From 4a50f773b71d0e8a51ad6de67769b9f973c7b065 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 21:26:04 +0000 Subject: [PATCH 1/5] fix(reporting-storage): upgrade 0.1.0b4/0.2.0b0 ledgers in place and pin the adcp storage contract adcp 10.0.0b4 added offering_id to the ledger column lists in adcp.reporting.ledger._record_rows, which adcp-reporting-storage reads against its own DDL. The only published storage release, 0.2.0b0, does not create the column, declares adcp>=9.0.0,<11, and so installs cleanly beside 10.0.0b4 and fails every reporting turn with UndefinedColumn (#1763). Storage side of the fix: - create_schema() now applies the versioned migrations (0001-0005) that are not yet recorded in adcp_schema_migrations inside its own transaction, under the reporting schema advisory lock, before its optional DDL. A ledger created by 0.1.0b4 or 0.2.0b0 is upgraded in place through either entry point: offering_id is added with '' for every existing account-level generation, obligation and status, the generation key and period index are widened, the managed-delivery foreign keys are re-pointed, and the applied schema versions are recorded. Re-running is a no-op; concurrent boots serialize. No evidence row is rewritten or dropped. - A storage contract number is declared on both sides (adcp_reporting_storage.STORAGE_CONTRACT = 2) and checked when a PostgreSQL ledger store is built: an adcp of another contract is refused at startup with REPORTING_SCHEMA_BEHIND naming both versions and the release to install, instead of failing on the first read. adcp releases without a declaration (10.0.0b4, b5) are classified by their column lists. - The dependency floor moves to adcp>=10.0.0b4,<11, the first release of contract 2, so the incompatible pair no longer resolves. tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py seeds ledgers with the actual published wheel pairs (adcp 9.3.1 + storage 0.1.0b4, adcp 10.0.0b2 + storage 0.2.0b0), reproduces the b4 failure on 0.2.0b0, and proves both upgrade paths keep every evidence row, read everything back, rebind the stored generations on restart, close the next period, record the version, re-run as a no-op, serialize concurrent boots, and produce a catalog identical to a fresh migration. Closes #1763 (storage side); the adcp-side startup check and the release gate follow in their own PRs. Co-Authored-By: Claude Fable 5.1 --- docs/reporting-ledger-migration.md | 32 ++ packages/adcp-reporting-storage/README.md | 31 +- .../adcp-reporting-storage/pyproject.toml | 12 +- .../src/adcp_reporting_storage/__init__.py | 8 + .../src/adcp_reporting_storage/contract.py | 93 ++++ .../src/adcp_reporting_storage/ledger/pg.py | 19 + .../migrations/__init__.py | 62 ++- .../reporting/_published_ledger_seed.py | 211 +++++++++ ..._reporting_storage_published_upgrade_pg.py | 411 ++++++++++++++++++ tests/test_reporting_storage_facade.py | 56 +++ 10 files changed, 911 insertions(+), 24 deletions(-) create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py create mode 100644 tests/conformance/reporting/_published_ledger_seed.py create mode 100644 tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md index 776b51791..87fd17d4f 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -281,3 +281,35 @@ history, not to authorize an unfrozen new publication. High-level users of `ProducerOfferings(currency=...)` keep their fixed-currency convenience; new obligations freeze that value. See [multi-account currency resolution and monetary validation](reporting-currency.md) for the resolver and type examples. + +## Upgrading a ledger created by adcp-reporting-storage 0.1.0b4 or 0.2.0b0 (#1763) + +adcp 10.0.0b4 widened the configuration generation key with `offering_id` +(AdCP 3.3 reporting subscriptions) and reads it from `reporting_configurations`, +`reporting_obligations` and `reporting_consumer_statuses`. The storage releases +before 0.2.0b1 do not create the column, so that pair fails every reporting +turn with `column "offering_id" does not exist`. From 0.2.0b1 the pair is +refused at startup instead (storage contract 1 versus 2; see +`packages/adcp-reporting-storage/README.md`), and both schema entry points +upgrade an existing ledger in place: + +- `PgReportingLedgerStore.create_schema()`, which `ReliableReportingService` + runs at startup when built with `store=`; +- `python -m adcp migrate`, which `postgres_storage(...)` requires to have run. + +The upgrade (migration `0002`) adds `offering_id TEXT NOT NULL DEFAULT ''` to +the widened tables, so every existing account-level generation, obligation and +status keeps its identity with `offering_id = ''`; rebuilds the generation +primary key and the obligation period index with the new column; and re-points +the managed-delivery foreign keys. No evidence row is rewritten or dropped. The +applied versions are recorded in `adcp_schema_migrations`, re-running is a +no-op, and concurrent boots serialize on the reporting schema advisory lock +inside one transaction. It is a stopped-worker upgrade: stop every worker of the +earlier release first, because their configuration writes name the +four-column key. + +`tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py` +seeds ledgers with the actual published wheel pairs (`adcp==9.3.1` with +`adcp-reporting-storage==0.1.0b4`, `adcp==10.0.0b2` with `0.2.0b0`) and proves +the retained evidence reads back unchanged, a restart rebinds the stored +generations, and the upgraded catalog is identical to a fresh migration. diff --git a/packages/adcp-reporting-storage/README.md b/packages/adcp-reporting-storage/README.md index d1b2a5c6c..ca1584896 100644 --- a/packages/adcp-reporting-storage/README.md +++ b/packages/adcp-reporting-storage/README.md @@ -53,10 +53,37 @@ client you build; the extras only pin the client libraries. This package implements the protocols of one `adcp` major and imports a few private helpers of `adcp.reporting` to do so (`adcp.reporting.ledger._record_rows`, -`_delivery_state`, `status_snapshot`). Its dependency range is therefore one -major (`adcp>=9.0.0,<10`), and a new `adcp` major means a new release here. +`_delivery_state`, `status_snapshot`). The column lists those helpers read are +the **storage contract**, numbered on both sides +(`adcp_reporting_storage.STORAGE_CONTRACT`, +`adcp.reporting.ledger.storage_contract.REPORTING_STORAGE_CONTRACT`) and bumped +together whenever the shared layout changes. The dependency floor is the first +`adcp` release of the contract this package implements, so an incompatible pair +does not resolve; if one is forced, building a PostgreSQL store refuses at +startup, naming both versions and the release to install. + +| adcp-reporting-storage | storage contract | adcp | +|---|---|---| +| 0.1.0b4 (SDK 9 line) | 1 | `>=9.0.0,<10` | +| 0.2.0b0 | 1 | `>=9.0.0,<11` (**do not pair with 10.0.0b4+**, [#1763](https://github.com/adcontextprotocol/adcp-client-python/issues/1763)) | +| 0.2.0b1+ | 2 | `>=10.0.0b4,<11` | + `adcp` never imports this package; a test in the main repository enforces that. +### Upgrading a ledger created by 0.1.0b4 or 0.2.0b0 + +Both `PgReportingLedgerStore.create_schema()` (what `ReliableReportingService` +runs at startup when given `store=`) and `python -m adcp migrate` upgrade an +existing ledger in place: `offering_id` is added to `reporting_configurations`, +`reporting_obligations`, `reporting_consumer_statuses` and the managed-delivery +tables, backfilled with `''` for every existing account-level generation, the +generation keys and unique indexes are widened, and the applied schema +versions are recorded in `adcp_schema_migrations`. No evidence row is rewritten +or dropped. The upgrade is transactional, serialized under the reporting schema +advisory lock, and a no-op when re-run. It is a **stopped-worker** upgrade: +stop every reporting worker of the earlier release first (its configuration +writes name the four-column key), run the upgrade, then start the new release. + ## Versioning 0.x: breaking changes may land in a minor release, each with a migration note in diff --git a/packages/adcp-reporting-storage/pyproject.toml b/packages/adcp-reporting-storage/pyproject.toml index e4dc08975..834479f7c 100644 --- a/packages/adcp-reporting-storage/pyproject.toml +++ b/packages/adcp-reporting-storage/pyproject.toml @@ -33,11 +33,13 @@ dependencies = [ # ``adcp.reporting`` helpers, so the range is one major: a new ``adcp`` # major means a new release here. # - # On next this range also admits adcp 10 pre-releases: next releases this - # distribution as its SDK 10 companion (0.2.0-beta.N), and the workspace - # must still install while pyproject.toml carries main's version. At - # ratification the range narrows to ``adcp>=10.0.0,<11`` (RELEASING.md). - "adcp>=9.0.0,<11", + # The floor is the first adcp release whose storage contract this package + # implements (``adcp_reporting_storage.contract``): 10.0.0b4 added + # ``offering_id`` to the ledger column lists, which 0.2.0b0's DDL lacked, so + # the pair installed cleanly and failed every reporting turn (#1763). Move + # the floor together with STORAGE_CONTRACT. At ratification the range + # narrows to ``adcp>=10.0.0,<11`` (RELEASING.md). + "adcp>=10.0.0b4,<11", # RFC 8785 canonical JSON for row bodies and object keys; the same pin as adcp. "rfc8785>=0.1.4", ] 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 e8b9f72bf..433363541 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py @@ -35,6 +35,10 @@ if TYPE_CHECKING: __version__: str + from adcp_reporting_storage.contract import STORAGE_CONTRACT as STORAGE_CONTRACT + from adcp_reporting_storage.contract import ( + require_adcp_contract as require_adcp_contract, + ) from adcp_reporting_storage.feed.pg import PgReportingFeedStore as PgReportingFeedStore from adcp_reporting_storage.inline_storage import ( PgReportingSealStore as PgReportingSealStore, @@ -214,6 +218,9 @@ #: Public name -> (module under this package, attribute). Resolved on first use. _LAZY_EXPORTS: dict[str, tuple[str, str]] = { + # The storage contract shared with adcp (see ``contract``) + "STORAGE_CONTRACT": ("contract", "STORAGE_CONTRACT"), + "require_adcp_contract": ("contract", "require_adcp_contract"), # Stores "PgReportingLedgerStore": ("ledger.pg", "PgReportingLedgerStore"), "PgReportingReconciliationStore": ("ledger.delivery_pg", "PgReportingReconciliationStore"), @@ -320,6 +327,7 @@ _LAZY_SUBMODULES = frozenset( { + "contract", "feed", "inline_storage", "ledger", diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py new file mode 100644 index 000000000..8fa2f3c8d --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py @@ -0,0 +1,93 @@ +"""The storage contract this distribution implements, checked against the installed ``adcp``. + +``adcp`` reads and writes the reporting ledger through private column lists +(``adcp.reporting.ledger._record_rows``) and this distribution ships the DDL +those lists are read against. The two must agree exactly: a column one side +knows and the other does not fails every reporting turn with +``UndefinedColumn`` (adcp-client-python#1763). The agreement is named by one +integer on each side, bumped together whenever the shared layout changes: + +* **1** — the SDK 9 layout and adcp 10.0.0b0–b3: a configuration generation is + ``(account_id, consumer_id, delivery_config_id, delivery_config_version)``. + Shipped by adcp-reporting-storage 0.1.0b4 and 0.2.0b0. +* **2** — adcp 10.0.0b4 and later: ``offering_id`` on configurations, + obligations and consumer statuses (AdCP 3.3 reporting subscriptions; schema + migration ``0002``). Shipped by adcp-reporting-storage 0.2.0b1 and later. + +:func:`require_adcp_contract` runs when a PostgreSQL ledger store is built, so +an incompatible pair fails at startup with the pair named and the fix spelled +out, instead of on the first ledger read. +""" + +from __future__ import annotations + +from importlib import import_module +from importlib.metadata import PackageNotFoundError, version + +__all__ = ["STORAGE_CONTRACT", "adcp_storage_contract", "require_adcp_contract"] + +#: The reporting storage contract this distribution's DDL and stores implement. +STORAGE_CONTRACT = 2 + + +def adcp_storage_contract() -> int: + """The storage contract the installed ``adcp`` reads the ledger with. + + adcp declares it from 10.0.0b6 (``adcp.reporting.ledger.storage_contract``). + Earlier releases are classified by the column list that distinguishes the + contracts: 10.0.0b4 and b5 select ``offering_id`` (contract 2); everything + before reads the four-column generation (contract 1). + """ + try: + declared = import_module("adcp.reporting.ledger.storage_contract") + except ImportError: + from adcp.reporting.ledger import _record_rows + + columns = [item.strip() for item in _record_rows._OBLIGATION_COLUMNS.split(",")] + return 2 if "offering_id" in columns else 1 + return int(declared.REPORTING_STORAGE_CONTRACT) + + +def _installed(distribution: str) -> str: + try: + return version(distribution) + except PackageNotFoundError: + return "unknown" + + +def require_adcp_contract() -> None: + """Refuse an ``adcp`` whose storage contract differs from this distribution's. + + Raises ``adcp.reporting.ReliableReportingConfigurationError`` (code + ``REPORTING_SCHEMA_BEHIND``) naming both installed versions, both contracts + and the release to install. + """ + expected = adcp_storage_contract() + if expected == STORAGE_CONTRACT: + return + from adcp.reporting._errors import ReliableReportingConfigurationError + + adcp_version = _installed("adcp") + storage_version = _installed("adcp-reporting-storage") + if expected > STORAGE_CONTRACT: + fix = ( + f"Install the adcp-reporting-storage release that implements contract {expected} " + f"(newer than {storage_version}), or adcp 10.0.0b4–10.0.0b5 with this one." + ) + else: + fix = ( + f"Install adcp 10.0.0b4 or later (storage contract {STORAGE_CONTRACT}), or " + "adcp-reporting-storage 0.2.0b0 with this adcp." + ) + message = ( + f"adcp {adcp_version} reads the reporting ledger with storage contract {expected}, " + f"but adcp-reporting-storage {storage_version} implements contract {STORAGE_CONTRACT}; " + f"the pair would fail every reporting turn. {fix}" + ) + try: + error = ReliableReportingConfigurationError( + message, code="REPORTING_SCHEMA_BEHIND", config_field="adcp-reporting-storage" + ) + except (TypeError, ValueError): # an adcp that predates the code or the keywords + error = ReliableReportingConfigurationError(message) + raise error diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 95be25a3d..fa6e57db3 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -154,6 +154,7 @@ validate_managed_revision_rows, validate_revision_currency, ) +from adcp_reporting_storage.contract import STORAGE_CONTRACT, require_adcp_contract from adcp_reporting_storage.ledger.row_storage import ( REPORTING_ROW_STORAGE_DDL_PATH, ReportingRowStorageOptionsV1, @@ -243,6 +244,9 @@ class PgReportingLedgerStore: """Durable reporting ledger over a caller-supplied connection pool.""" is_durable: ClassVar[bool] = True + #: The reporting storage contract this store's DDL implements; adcp refuses a + #: store whose contract differs from its own (``adcp_reporting_storage.contract``). + storage_contract: ClassVar[int] = STORAGE_CONTRACT def __init__( self, @@ -255,6 +259,8 @@ def __init__( ) -> None: if not PG_AVAILABLE: raise ImportError(_INSTALL_HINT) + # Fail at construction, naming the pair, rather than on the first read. + require_adcp_contract() self._pool = pool # Opt-in chunked row storage. Without it new revisions keep their rows in # reporting_revision_rows; reads handle both forms either way. @@ -472,6 +478,15 @@ async def publish(key: str, sealed: SealedSlice) -> SealedSlice: async def create_schema(self) -> None: """Create or upgrade the ledger atomically, serializing concurrent boots. + Applies the versioned migrations (``adcp_reporting_storage.migrations``) + that are not yet recorded in ``adcp_schema_migrations``, then this + store's optional DDL (change feed, row storage), all in one transaction + under the reporting schema lock. A ledger created by adcp-reporting-storage + 0.1.0b4 or 0.2.0b0 is upgraded in place: ``offering_id`` is added with + ``''`` for every existing account-level generation, and its schema + version is recorded, so ``python -m adcp migrate --status`` and + ``postgres_storage`` read it as current. Re-running is a no-op. + The bootstrap takes a transaction-scoped schema lock before any DDL. Keep the migration in that transaction, including with an autocommit pool, so a second process cannot observe a partially upgraded schema. @@ -666,8 +681,12 @@ async def retire_legacy_inline_rows(self) -> bool: async def _create_schema_on(self, connection: Any) -> None: from adcp_reporting_storage.migration import require_owned_schema_or_empty + from adcp_reporting_storage.migrations import apply_pending await require_owned_schema_or_empty(connection) + # The versioned files first, recorded as applied; the idempotent DDL + # below then only re-verifies what they installed. + await apply_pending(connection) for path in ( _DDL_PATH, _ACCOUNT_GENERATIONS_DDL_PATH, 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 index ab5250ca8..62fd77ab0 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py @@ -18,6 +18,9 @@ Every migration runs in its own transaction under the reporting schema lock, so concurrent runners serialize and a failure commits nothing. +``PgReportingLedgerStore.create_schema()`` applies the same files, recorded the +same way, inside its own bootstrap transaction (:func:`apply_pending`), so a +ledger created or upgraded by either entry point carries its schema version. """ from __future__ import annotations @@ -34,6 +37,7 @@ "Migration", "MigrationStatus", "applied_migrations", + "apply_pending", "migrate", "migration_status", "migrations", @@ -179,29 +183,55 @@ async def migrate(pool: Any, *, dry_run: bool = False) -> list[Migration]: 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) + if await _apply(connection, migration): + applied.append(migration) return applied +async def apply_pending(connection: Any) -> list[Migration]: + """Apply every pending migration on ``connection`` inside the caller's transaction. + + The bootstrap path of ``create_schema()``: the caller holds one transaction + for the whole upgrade, so every migration commits with it or none does. + Takes the reporting schema lock first, like :func:`migrate`, so a booting + worker and ``python -m adcp migrate`` serialize instead of racing. + """ + await connection.execute(_LOCK) + await connection.execute(_TABLE) + applied = await applied_migrations(connection) + result: list[Migration] = [] + for migration in migrations(): + if migration.version in applied: + continue + if await _apply(connection, migration): + result.append(migration) + return result + + +async def _apply(connection: Any, migration: Migration) -> bool: + """Run one migration and record it; ``False`` when a concurrent runner already had.""" + from adcp_reporting_storage.migration import require_owned_schema_or_empty + + await connection.execute(_LOCK) + await connection.execute(_TABLE) + if migration.version in await applied_migrations(connection): + return False # 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), + ) + return True + + def describe(statuses: Sequence[MigrationStatus]) -> list[str]: """``--status`` lines: one per migration, flagging stopped-worker upgrades.""" lines: list[str] = [] @@ -211,9 +241,7 @@ def describe(statuses: Sequence[MigrationStatus]) -> list[str]: 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 "" + else " [stopped-worker upgrade]" if item.stopped_worker else "" ) lines.append(f"{item.version} {item.name}: {state}{note}") return lines diff --git a/tests/conformance/reporting/_published_ledger_seed.py b/tests/conformance/reporting/_published_ledger_seed.py new file mode 100644 index 000000000..3b2c1912a --- /dev/null +++ b/tests/conformance/reporting/_published_ledger_seed.py @@ -0,0 +1,211 @@ +"""Run under an installed adcp + adcp-reporting-storage pair, outside the checkout. + +Reads one JSON settings object on stdin and prints one JSON result. ``seed`` +creates the ledger with the installed pair's own ``create_schema()`` and writes +two account-level generations with obligations, revisions and rows, the way a +seller on that release did; ``read`` reads them back; ``upgrade`` runs the +installed storage's ``create_schema()`` first. Every action returns an ``image`` +of every ``reporting_*``/``adcp_reporting_*`` table so the test can prove no +retained evidence changed. Only APIs common to adcp 9.3.1 and 10.0.0b2+ are used. +""" + +import asyncio +import json +import sys +from datetime import datetime, timedelta, timezone + +ACTIVATED_AT = datetime(2026, 11, 1, 0, 20, tzinfo=timezone.utc) +ACCOUNTS = ("account-redacted", "account-b") +CONSUMER = "buyer-1" +VOLATILE_TABLES = {"adcp_schema_migrations"} + + +def _configuration(account_id): + from adcp.reporting.ledger import ( + ReportingConfiguration, + ReportingDefinitionBinding, + ReportingScheduleSpec, + ) + + return ReportingConfiguration( + delivery_config_id="gam-delivery", + delivery_config_version=1, + account_id=account_id, + consumer_id=CONSUMER, + report_definition_id="PAID_MEDIA_DAILY_V1", + reporting_profile="paid_media_delivery", + feed_purpose="analytics", + schedule=ReportingScheduleSpec( + period_duration="PT1H", delivery_sla="PT10M", alignment="utc" + ), + required_finality="snapshot", + activated_at=ACTIVATED_AT, + media_buy_ids=("media-buy-redacted",), + definition=ReportingDefinitionBinding( + report_definition_uri="https://contracts.example.test/reporting/paid-media-daily-v1", + report_definition_sha256="a" * 64, + schema_version="1.0.0", + schema_uri="https://contracts.example.test/reporting/paid-media-daily-v1/schema", + schema_sha256="b" * 64, + ), + ) + + +def _obligation(configuration, ordinal): + from adcp.reporting.ledger import ReportingObligationRecord, derive_period + from adcp.reporting.ledger.models import first_ordinal_after + + schedule = configuration.schedule + base = first_ordinal_after(schedule, account_timezone="UTC", activated_at=ACTIVATED_AT) + period = derive_period(schedule, account_timezone="UTC", ordinal=base + ordinal) + return ReportingObligationRecord( + reporting_obligation_id=f"rpo_{configuration.account_id}_{ordinal}", + account_id=configuration.account_id, + consumer_id=configuration.consumer_id, + delivery_config_id=configuration.delivery_config_id, + delivery_config_version=configuration.delivery_config_version, + report_definition_id=configuration.report_definition_id, + reporting_profile=configuration.reporting_profile, + feed_purpose=configuration.feed_purpose, + period=period, + scope_resolved_at=period.end, + media_buy_ids=configuration.media_buy_ids, + required_finality=configuration.required_finality, + automated_recovery_deadline_at=period.expected_at + timedelta(hours=6), + schedule=schedule, + definition=configuration.definition, + created_at=period.end, + currency="USD", + ) + + +def _revision(obligation, revision_id): + from adcp.reporting.ledger import ReportingRevisionRecord, revision_content_sha256 + + rows = [ + { + "media_buy_id": "media-buy-redacted", + "campaign_id": "campaign-redacted-1", + "impressions": 10, + "spend": "1.25", + } + ] + totals = (("impressions", "10"), ("spend", "1.25")) + record = ReportingRevisionRecord( + reporting_revision_id=revision_id, + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + finality="snapshot", + revision_content_sha256=revision_content_sha256( + reporting_revision_id=revision_id, + row_count=len(rows), + control_totals=totals, + reporting_rows=rows, + ), + row_count=len(rows), + control_totals=totals, + observed_at=obligation.period.end, + data_through=obligation.period.end, + created_at=obligation.period.expected_at, + ) + return record, rows + + +async def image(connection): + """Every reporting table's rows as JSON, in a stable order.""" + tables = await ( + await connection.execute( + "SELECT tablename FROM pg_tables WHERE schemaname = current_schema()" + " AND (starts_with(tablename, 'reporting_') OR starts_with(tablename, 'adcp_'))" + " ORDER BY tablename" + ) + ).fetchall() + result = {} + for (table,) in tables: + if table in VOLATILE_TABLES: + continue + rows = await ( + await connection.execute( + f"SELECT to_jsonb(t) FROM {table} t ORDER BY to_jsonb(t)::text" # noqa: S608 + ) + ).fetchall() + result[table] = [row[0] for row in rows] + return result + + +async def readback(store): + from adcp.reporting.ledger import ReportingStatusCaller + + result = {"configurations": [], "obligations": [], "revisions": [], "rows": {}} + for account_id in ACCOUNTS: + caller = ReportingStatusCaller(account_id=account_id, consumer_id=CONSUMER) + for configuration in await store.list_configurations(caller=caller): + result["configurations"].append( + [ + configuration.account_id, + configuration.consumer_id, + configuration.delivery_config_id, + configuration.delivery_config_version, + ] + ) + for ordinal in range(2): + seeded = _obligation(_configuration(account_id), ordinal) + found = await store.find_obligation( + account_id=account_id, + consumer_id=CONSUMER, + delivery_config_id="gam-delivery", + delivery_config_version=1, + period_start=seeded.period.start, + period_end=seeded.period.end, + ) + result["obligations"].append(found.reporting_obligation_id if found else None) + if found is None: + continue + for revision in await store.list_revisions( + account_id=account_id, reporting_obligation_id=found.reporting_obligation_id + ): + result["revisions"].append(revision.reporting_revision_id) + page = await store.read_revision_rows( + account_id=account_id, reporting_revision_id=revision.reporting_revision_id + ) + result["rows"][revision.reporting_revision_id] = list(page.rows) + return result + + +async def main(settings): + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from psycopg_pool import AsyncConnectionPool + + async with AsyncConnectionPool( + settings["conninfo"], kwargs=settings["kwargs"], min_size=1, max_size=4, open=False + ) as pool: + await pool.open() + store = PgReportingLedgerStore(pool=pool) + result = {"action": settings["action"]} + try: + if settings["action"] in {"seed", "upgrade"}: + await store.create_schema() + if settings["action"] == "seed": + for account_id in ACCOUNTS: + configuration = _configuration(account_id) + await store.put_configuration(configuration) + for ordinal in range(2): + obligation = await store.commit_obligation( + _obligation(configuration, ordinal) + ) + revision, rows = await _async_revision(obligation, account_id, ordinal) + await store.commit_revision(revision, rows) + result["readback"] = await readback(store) + except Exception as error: # reported, so the test can assert on the repro + result["error"] = f"{type(error).__name__}: {error}" + async with pool.connection() as connection: + result["image"] = await image(connection) + return result + + +async def _async_revision(obligation, account_id, ordinal): + return _revision(obligation, f"rpr_{account_id}_{ordinal}") + + +if __name__ == "__main__": + print(json.dumps(asyncio.run(main(json.loads(sys.stdin.read()))), default=str)) diff --git a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py new file mode 100644 index 000000000..522baed80 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py @@ -0,0 +1,411 @@ +"""Ledgers created by the published adcp-reporting-storage wheels upgrade in place. + +adcp 10.0.0b4 added ``offering_id`` to the ledger column lists; the only +published storage release, 0.2.0b0, did not create the column, and the pair +installed cleanly and failed every reporting turn (#1763). A seller's +production ledger created by 0.1.0b4 or 0.2.0b0 cannot be recreated, so these +tests seed ledgers with the *actual published wheels* (``adcp==9.3.1`` with +``adcp-reporting-storage==0.1.0b4``, ``adcp==10.0.0b2`` with ``0.2.0b0``), +with generations, obligations, revisions and rows, then upgrade them with this +checkout through both entry points and prove: + +* no retained evidence row changes beyond the added ``offering_id = ''``; +* every record reads back and a restart rebinds the stored generations; +* the upgraded catalog is identical to a freshly migrated one; +* the applied schema version is recorded, re-running is a no-op, and + concurrent boots serialize; +* ``adcp==10.0.0b4`` with *this* storage, the pair a seller upgrades to, + upgrades a 0.2.0b0 ledger, where the published 0.2.0b0 fails. + +Requires ``ADCP_PG_TEST_URL`` and PyPI; skips otherwise. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import pytest + +from adcp.reporting.ledger import ReportingConfigurationGenerationKey + +from ._generation_support import isolated_reporting_pool + +pytestmark = pytest.mark.skipif( + not os.environ.get("ADCP_PG_TEST_URL"), reason="ADCP_PG_TEST_URL not set" +) + +ROOT = Path(__file__).resolve().parents[3] +STORAGE_ROOT = ROOT / "packages/adcp-reporting-storage" +SEED = Path(__file__).with_name("_published_ledger_seed.py") + +#: Published pairs that created ledgers in the field: the SDK 9 line and the +#: last SDK 10 beta before the contract changed. +PUBLISHED_PAIRS = {"0.1.0b4": "9.3.1", "0.2.0b0": "10.0.0b2"} +#: The first adcp whose column lists carry ``offering_id`` (storage contract 2). +FIRST_CONTRACT_2_ADCP = "10.0.0b4" +#: Tables the upgrade widens; every existing row gains ``offering_id = ''``. +WIDENED = { + "reporting_configurations", + "reporting_obligations", + "reporting_consumer_statuses", + "reporting_reconciliation_records", + "reporting_materializer_discovery", + "reporting_status_scope_checkpoints", + "reporting_production_generations", +} +ACCOUNTS = ("account-redacted", "account-b") +SEEDED_KEYS = { + ReportingConfigurationGenerationKey(account, "buyer-1", "gam-delivery", 1) + for account in ACCOUNTS +} +RESTART_NOW = datetime(2026, 11, 1, 4, 30, tzinfo=timezone.utc) + +CATALOG = """ +SELECT 'column ' || table_name || '.' || column_name || ' ' || data_type + || ' null=' || is_nullable || ' default=' || coalesce(column_default, '') + || ' collation=' || coalesce(collation_name, '') +FROM information_schema.columns WHERE table_schema = current_schema() +UNION ALL +SELECT 'constraint ' || conrelid::regclass || ' ' || conname || ' ' || pg_get_constraintdef(oid) +FROM pg_constraint WHERE connamespace = current_schema()::regnamespace +UNION ALL +SELECT 'index ' || tablename || ' ' || indexname || ' ' + || regexp_replace(indexdef, ' ON [^ ]+[.]', ' ON ') +FROM pg_indexes WHERE schemaname = current_schema() +UNION ALL +SELECT 'trigger ' || tgrelid::regclass || ' ' || tgname FROM pg_trigger +WHERE NOT tgisinternal + AND tgrelid IN (SELECT oid FROM pg_class WHERE relnamespace = current_schema()::regnamespace) +""" + + +# -- installed pairs ------------------------------------------------------- + + +def _install(python: Path, *requirements: str) -> None: + if shutil.which("uv"): + command = ["uv", "pip", "install", "--quiet", "--python", str(python), *requirements] + else: + command = [str(python), "-m", "pip", "install", "--quiet", *requirements] + completed = subprocess.run(command, capture_output=True, text=True, timeout=600) + if completed.returncode != 0: + pytest.skip(f"cannot install {' '.join(requirements)} (PyPI unreachable?)") + + +def _environment(root: Path, name: str, *requirements: str) -> Path: + """A private interpreter with ``requirements`` installed from PyPI.""" + target = root / name + if shutil.which("uv"): + command = ["uv", "venv", "--quiet", "--python", sys.executable, str(target)] + else: + command = [sys.executable, "-m", "venv", str(target)] + subprocess.run(command, check=True, capture_output=True, timeout=120) + python = target / "bin" / "python" + _install(python, *requirements) + return python + + +def _run(python: Path, settings: dict[str, Any], *, cwd: Path) -> dict[str, Any]: + """Run the seed script under ``python`` outside the checkout, isolated from it.""" + script = cwd / "published_ledger_seed.py" + shutil.copy2(SEED, script) + environment = {key: value for key, value in os.environ.items() if key != "PYTHONPATH"} + completed = subprocess.run( + [str(python), "-I", str(script)], + input=json.dumps(settings), + capture_output=True, + text=True, + cwd=cwd, + env=environment, + timeout=180, + ) + assert completed.returncode == 0, completed.stderr[-4000:] + return json.loads(completed.stdout) + + +@pytest.fixture(scope="module") +def workspace(tmp_path_factory: pytest.TempPathFactory) -> Path: + return tmp_path_factory.mktemp("published-pairs") + + +@pytest.fixture(scope="module", params=sorted(PUBLISHED_PAIRS)) +def published(request: pytest.FixtureRequest, workspace: Path) -> tuple[str, Path]: + """``(storage version, python)`` for a published pair, as a seller installed it.""" + storage = request.param + python = _environment( + workspace, + f"pair-{storage}", + f"adcp=={PUBLISHED_PAIRS[storage]}", + f"adcp-reporting-storage[pg]=={storage}", + ) + return storage, python + + +@pytest.fixture(scope="module") +def b4_with_published_storage(workspace: Path) -> Path: + """The failing pair of #1763: adcp 10.0.0b4 with the published 0.2.0b0.""" + return _environment( + workspace, + "b4-published", + f"adcp=={FIRST_CONTRACT_2_ADCP}", + "adcp-reporting-storage[pg]==0.2.0b0", + ) + + +@pytest.fixture(scope="module") +def b4_with_this_storage(workspace: Path) -> Path: + """The pair a seller upgrades to: adcp 10.0.0b4 with this checkout's storage.""" + python = _environment( + workspace, + "b4-this", + f"adcp=={FIRST_CONTRACT_2_ADCP}", + "adcp-reporting-storage[pg]==0.2.0b0", + ) + _install(python, "--no-deps", str(STORAGE_ROOT)) + return python + + +def _settings(pool: Any, action: str) -> dict[str, Any]: + return {"conninfo": pool.conninfo, "kwargs": pool.kwargs, "action": action} + + +async def _seed(python: Path, pool: Any, cwd: Path) -> dict[str, Any]: + result = await asyncio.to_thread(_run, python, _settings(pool, "seed"), cwd=cwd) + assert "error" not in result, result["error"] + assert sorted(result["readback"]["revisions"]) == sorted( + f"rpr_{account}_{ordinal}" for account in ACCOUNTS for ordinal in range(2) + ) + return result + + +# -- assertions ------------------------------------------------------------ + + +def _normalized(image: dict[str, list[dict[str, Any]]]) -> dict[str, list[str]]: + return { + table: sorted(json.dumps(row, sort_keys=True) for row in rows) + for table, rows in image.items() + } + + +def assert_evidence_unchanged(before: dict[str, Any], after: dict[str, Any]) -> None: + """Every pre-upgrade row survives byte-for-byte, plus ``offering_id = ''`` where widened.""" + expected = { + table: [{**row, "offering_id": ""} if table in WIDENED else row for row in rows] + for table, rows in before.items() + } + for table, rows in _normalized(expected).items(): + assert _normalized({table: after[table]})[table] == rows, table + for table in set(after) - set(before): + assert after[table] == [], f"the upgrade wrote rows into new table {table}" + + +async def _versions(pool: Any) -> list[str]: + from adcp_reporting_storage.migrations import migration_status, migrations + + statuses = await migration_status(pool) + assert [item.migration.version for item in statuses] == [m.version for m in migrations()] + return [item.migration.version for item in statuses if not item.pending] + + +async def _readback_here(pool: Any) -> dict[str, Any]: + """Read the ledger with this checkout, through the same calls the seed used.""" + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + from tests.conformance.reporting import _published_ledger_seed as seed + + return await seed.readback(PgReportingLedgerStore(pool=pool)) + + +async def _image_here(pool: Any) -> dict[str, Any]: + from tests.conformance.reporting import _published_ledger_seed as seed + + async with pool.connection() as connection: + return await seed.image(connection) + + +async def _catalog(pool: Any) -> list[str]: + async with pool.connection() as connection: + rows = await (await connection.execute(CATALOG)).fetchall() + return sorted(row[0] for row in rows) + + +# -- tests ----------------------------------------------------------------- + + +async def test_create_schema_upgrades_a_published_ledger_in_place( + published: tuple[str, Path], workspace: Path +) -> None: + """The ``store=`` startup path (``ReliableReportingService`` calls ``create_schema``).""" + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + from adcp.storage.postgres import postgres_storage + + storage, python = published + async with isolated_reporting_pool(autocommit=True) as pool: + seeded = await _seed(python, pool, workspace) + assert await _versions(pool) == [], f"{storage} recorded no schema version" + assert "offering_id" not in seeded["image"]["reporting_obligations"][0] + + store = PgReportingLedgerStore(pool=pool) + await store.create_schema() + + assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + after = await _image_here(pool) + assert_evidence_unchanged(seeded["image"], after) + assert await _readback_here(pool) == seeded["readback"] + await postgres_storage(pool=pool).check_schema() # the migrations-managed path agrees + + await store.create_schema() # re-runnable: nothing changes + assert await _image_here(pool) == after + assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + + +async def test_migrate_upgrades_a_published_ledger_to_the_fresh_catalog( + published: tuple[str, Path], workspace: Path +) -> None: + """``python -m adcp migrate`` on the seller's ledger yields a freshly migrated catalog.""" + from adcp_reporting_storage.migrations import migrate + + _, python = published + async with isolated_reporting_pool(autocommit=True) as pool: + seeded = await _seed(python, pool, workspace) + applied = await migrate(pool) + assert [item.version for item in applied] == ["0001", "0002", "0003", "0004", "0005"] + assert await migrate(pool) == [] + assert_evidence_unchanged(seeded["image"], await _image_here(pool)) + assert await _readback_here(pool) == seeded["readback"] + upgraded = await _catalog(pool) + async with isolated_reporting_pool(autocommit=True) as fresh: + await migrate(fresh) + assert upgraded == await _catalog(fresh) + + +async def test_restart_rebinds_stored_generations_and_closes_the_next_period( + published: tuple[str, Path], workspace: Path +) -> None: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + from adcp.reporting.fixtures import redacted_capabilities + from adcp.reporting.service import ReliableReportingService + from adcp.reporting.testing import ScriptedReportingAdapter + from tests.conformance.reporting import _published_ledger_seed as seed + from tests.test_reliable_reporting_service import _account_context, _rows + + _, python = published + async with isolated_reporting_pool(autocommit=True) as pool: + seeded = await _seed(python, pool, workspace) + service = ReliableReportingService( + store=PgReportingLedgerStore(pool=pool, clock=lambda: RESTART_NOW), + clock=lambda: RESTART_NOW, + account_context=_account_context, + ) + service.sources.register( + "gam", ScriptedReportingAdapter(redacted_capabilities(), [_rows(10)] * 8) + ) + try: + await service.start() # create_schema(): the in-place upgrade + assert service.initialization_errors == {} + turn = await service.run_worker(now=RESTART_NOW) + finally: + await service.close() + assert turn.configuration_errors == {} + assert set(turn.configurations) == SEEDED_KEYS + store = PgReportingLedgerStore(pool=pool) + for account in ACCOUNTS: + configuration = seed._configuration(account) + # Periods 01:00 and 02:00 were seeded; the restart closes 03:00-04:00. + next_period = seed._obligation(configuration, 2).period + closed = await store.find_obligation( + account_id=account, + consumer_id="buyer-1", + delivery_config_id="gam-delivery", + delivery_config_version=1, + period_start=next_period.start, + period_end=next_period.end, + ) + assert closed is not None, f"{account}: no obligation for the next period" + revisions = await store.list_revisions( + account_id=account, reporting_obligation_id=closed.reporting_obligation_id + ) + assert [item.finality for item in revisions] == ["snapshot"] + # The seeded evidence is untouched by the turn. + after = await _image_here(pool) + for table in ("reporting_revisions", "reporting_revision_rows"): + seeded_rows = _normalized(seeded["image"])[table] + assert set(seeded_rows) <= set(_normalized(after)[table]), table + # The turn may re-observe the seeded snapshot periods (new revisions) + # but every seeded revision still reads. + assert set(seeded["readback"]["revisions"]) <= set( + (await _readback_here(pool))["revisions"] + ) + + +async def test_concurrent_boots_and_migrate_serialize_on_a_published_ledger( + published: tuple[str, Path], workspace: Path +) -> None: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from adcp_reporting_storage.migrations import migrate + from psycopg_pool import AsyncConnectionPool + + _, python = published + async with isolated_reporting_pool(autocommit=True) as pool: + seeded = await _seed(python, pool, workspace) + pools = [ + AsyncConnectionPool( + pool.conninfo, kwargs=pool.kwargs, min_size=1, max_size=2, open=False + ) + for _ in range(4) + ] + for item in pools: + await item.open(wait=True) + try: + await asyncio.gather( + PgReportingLedgerStore(pool=pools[0]).create_schema(), + migrate(pools[1]), + PgReportingLedgerStore(pool=pools[2]).create_schema(), + migrate(pools[3]), + ) + finally: + for item in pools: + await item.close() + assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + async with pool.connection() as connection: + row = await ( + await connection.execute("SELECT count(*) FROM adcp_schema_migrations") + ).fetchone() + assert row[0] == 5 + assert_evidence_unchanged(seeded["image"], await _image_here(pool)) + + +async def test_adcp_b4_fails_on_the_published_storage_and_upgrades_with_this_one( + published: tuple[str, Path], + b4_with_published_storage: Path, + b4_with_this_storage: Path, + workspace: Path, +) -> None: + """The reported failure, then the pair the fix ships: adcp 10.0.0b4 + this storage.""" + storage, python = published + if storage != "0.2.0b0": + pytest.skip("the 10.0.0b4 pairing concerns the 0.2.0b0 ledger") + async with isolated_reporting_pool(autocommit=True) as pool: + seeded = await _seed(python, pool, workspace) + failed = await asyncio.to_thread( + _run, b4_with_published_storage, _settings(pool, "read"), cwd=workspace + ) + assert "offering_id" in failed.get("error", ""), failed # #1763, reproduced + upgraded = await asyncio.to_thread( + _run, b4_with_this_storage, _settings(pool, "upgrade"), cwd=workspace + ) + assert "error" not in upgraded, upgraded["error"] + assert upgraded["readback"] == seeded["readback"] + assert_evidence_unchanged(seeded["image"], upgraded["image"]) + assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] diff --git a/tests/test_reporting_storage_facade.py b/tests/test_reporting_storage_facade.py index 9702434cc..c199a6065 100644 --- a/tests/test_reporting_storage_facade.py +++ b/tests/test_reporting_storage_facade.py @@ -163,3 +163,59 @@ def test_postgres_storage_resolves_through_the_entry_point() -> None: from adcp.storage import _plugin assert _plugin("adcp.storage", "postgres") is direct + + +# -- the storage contract shared with adcp-reporting-storage (#1763) --------- + + +def test_storage_declares_the_contract_adcp_reads_with() -> None: + pytest.importorskip("adcp_reporting_storage") + import adcp_reporting_storage + from adcp_reporting_storage.contract import STORAGE_CONTRACT, adcp_storage_contract + + assert adcp_reporting_storage.STORAGE_CONTRACT == STORAGE_CONTRACT == 2 + assert adcp_storage_contract() == STORAGE_CONTRACT + adcp_reporting_storage.require_adcp_contract() # the workspace pair agrees + + +def test_storage_classifies_an_adcp_without_a_declared_contract_by_its_columns( + monkeypatch, +) -> None: + """adcp 10.0.0b4/b5 declare nothing; their column lists name the contract.""" + pytest.importorskip("adcp_reporting_storage") + import sys + + from adcp_reporting_storage.contract import adcp_storage_contract + + from adcp.reporting.ledger import _record_rows + + monkeypatch.setitem(sys.modules, "adcp.reporting.ledger.storage_contract", None) + assert adcp_storage_contract() == 2 + monkeypatch.setattr( + _record_rows, + "_OBLIGATION_COLUMNS", + _record_rows._OBLIGATION_COLUMNS.replace(", offering_id", ""), + ) + assert adcp_storage_contract() == 1 # 9.x and 10.0.0b0-b3 + + +def test_a_postgres_store_refuses_an_adcp_of_another_contract(monkeypatch) -> None: + pytest.importorskip("adcp_reporting_storage") + pytest.importorskip("psycopg_pool") + from adcp_reporting_storage import contract + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + from adcp.reporting import ReliableReportingConfigurationError + + assert PgReportingLedgerStore.storage_contract == contract.STORAGE_CONTRACT + monkeypatch.setattr(contract, "adcp_storage_contract", lambda: 1) + with pytest.raises(ReliableReportingConfigurationError) as refused: + PgReportingLedgerStore(pool=object()) + error = refused.value + assert error.code == "REPORTING_SCHEMA_BEHIND" and error.recovery == "correctable" + assert error.config_field == "adcp-reporting-storage" + assert "storage contract 1" in str(error) and "implements contract 2" in str(error) + assert "Install adcp 10.0.0b4 or later" in str(error) + monkeypatch.setattr(contract, "adcp_storage_contract", lambda: 3) + with pytest.raises(ReliableReportingConfigurationError, match="newer than"): + PgReportingLedgerStore(pool=object()) From 60aa18b166345ab84d68f01967e8c73adbfbd213 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 21:46:53 +0000 Subject: [PATCH 2/5] fix(reporting-storage): bound create_schema to the ledger's own migrations create_schema applies and records the migrations made of the DDL this store owns (0001-0003) and stops there: 0004 (subscription state) and 0005 (inline storage) create reporting_* tables of other stores, and the A/B catalog-identity suites assert that an older binary's installed object set is exactly its own manifest. Those two stay pending for python -m adcp migrate, which postgres_storage requires anyway. Also: the contract module imports ReliableReportingConfigurationError through the public adcp.reporting name, the public API snapshot records STORAGE_CONTRACT and require_adcp_contract, and the release configuration test pins next's storage floor to the first adcp of its storage contract. Co-Authored-By: Claude Fable 5.1 --- docs/reporting-ledger-migration.md | 8 +++-- packages/adcp-reporting-storage/README.md | 6 ++-- .../src/adcp_reporting_storage/contract.py | 2 +- .../src/adcp_reporting_storage/ledger/pg.py | 36 ++++++++++++------- .../migrations/__init__.py | 21 ++++++++--- ..._reporting_storage_published_upgrade_pg.py | 26 ++++++++++---- tests/fixtures/public_api_snapshot.json | 2 ++ tests/test_release_configuration.py | 15 ++++---- 8 files changed, 80 insertions(+), 36 deletions(-) diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md index 87fd17d4f..765a9a6d2 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -302,9 +302,11 @@ the widened tables, so every existing account-level generation, obligation and status keeps its identity with `offering_id = ''`; rebuilds the generation primary key and the obligation period index with the new column; and re-points the managed-delivery foreign keys. No evidence row is rewritten or dropped. The -applied versions are recorded in `adcp_schema_migrations`, re-running is a -no-op, and concurrent boots serialize on the reporting schema advisory lock -inside one transaction. It is a stopped-worker upgrade: stop every worker of the +applied versions are recorded in `adcp_schema_migrations` (`create_schema()` +applies the ledger's own `0001`–`0003`; `python -m adcp migrate` also applies +`0004`–`0005`, which only `postgres_storage` needs), re-running is a no-op, +and concurrent boots serialize on the reporting schema advisory lock inside +one transaction. It is a stopped-worker upgrade: stop every worker of the earlier release first, because their configuration writes name the four-column key. diff --git a/packages/adcp-reporting-storage/README.md b/packages/adcp-reporting-storage/README.md index ca1584896..dd28ad4d3 100644 --- a/packages/adcp-reporting-storage/README.md +++ b/packages/adcp-reporting-storage/README.md @@ -78,8 +78,10 @@ existing ledger in place: `offering_id` is added to `reporting_configurations`, `reporting_obligations`, `reporting_consumer_statuses` and the managed-delivery tables, backfilled with `''` for every existing account-level generation, the generation keys and unique indexes are widened, and the applied schema -versions are recorded in `adcp_schema_migrations`. No evidence row is rewritten -or dropped. The upgrade is transactional, serialized under the reporting schema +versions are recorded in `adcp_schema_migrations` (`create_schema()` applies +and records the ledger's own `0001`–`0003`; `python -m adcp migrate` also +applies `0004`–`0005`, which only `postgres_storage` needs). No evidence row is +rewritten or dropped. The upgrade is transactional, serialized under the reporting schema advisory lock, and a no-op when re-run. It is a **stopped-worker** upgrade: stop every reporting worker of the earlier release first (its configuration writes name the four-column key), run the upgrade, then start the new release. diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py index 8fa2f3c8d..08c291ca9 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/contract.py @@ -65,7 +65,7 @@ def require_adcp_contract() -> None: expected = adcp_storage_contract() if expected == STORAGE_CONTRACT: return - from adcp.reporting._errors import ReliableReportingConfigurationError + from adcp.reporting import ReliableReportingConfigurationError adcp_version = _installed("adcp") storage_version = _installed("adcp-reporting-storage") diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index fa6e57db3..119cc0011 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -478,14 +478,16 @@ async def publish(key: str, sealed: SealedSlice) -> SealedSlice: async def create_schema(self) -> None: """Create or upgrade the ledger atomically, serializing concurrent boots. - Applies the versioned migrations (``adcp_reporting_storage.migrations``) - that are not yet recorded in ``adcp_schema_migrations``, then this - store's optional DDL (change feed, row storage), all in one transaction - under the reporting schema lock. A ledger created by adcp-reporting-storage + Applies the versioned migrations made of this store's DDL + (``adcp_reporting_storage.migrations`` ``0001``–``0003``) that are not + yet recorded in ``adcp_schema_migrations``, then this store's optional + DDL (change feed, row storage), all in one transaction under the + reporting schema lock. A ledger created by adcp-reporting-storage 0.1.0b4 or 0.2.0b0 is upgraded in place: ``offering_id`` is added with - ``''`` for every existing account-level generation, and its schema - version is recorded, so ``python -m adcp migrate --status`` and - ``postgres_storage`` read it as current. Re-running is a no-op. + ``''`` for every existing account-level generation, and the applied + versions are recorded, so ``python -m adcp migrate --status`` reads + them. Re-running is a no-op. ``postgres_storage`` additionally needs + ``0004``/``0005``, which only ``python -m adcp migrate`` applies. The bootstrap takes a transaction-scoped schema lock before any DDL. Keep the migration in that transaction, including with an autocommit @@ -684,10 +686,7 @@ async def _create_schema_on(self, connection: Any) -> None: from adcp_reporting_storage.migrations import apply_pending await require_owned_schema_or_empty(connection) - # The versioned files first, recorded as applied; the idempotent DDL - # below then only re-verifies what they installed. - await apply_pending(connection) - for path in ( + owned = ( _DDL_PATH, _ACCOUNT_GENERATIONS_DDL_PATH, _CURRENCY_DDL_PATH, @@ -696,7 +695,20 @@ async def _create_schema_on(self, connection: Any) -> None: _ACTIVITY_DDL_PATH, Path(__file__).with_name("reporting_provisional_observations.sql"), Path(__file__).with_name("reporting_caller_ownership.sql"), - ): + ) + # The versioned migrations made of this store's own DDL first (0001-0003), + # recorded as applied, so a 0.1.0b4/0.2.0b0 ledger is upgraded in place + # and carries its version; the idempotent files below then only + # re-verify what they installed. Migrations of other stores' tables + # (0004 subscription state, 0005 inline storage) stay pending for + # python -m adcp migrate: the catalog contract of this store's older + # binaries is exactly its own reporting_* objects. + package = Path(__file__).resolve().parents[1] + await apply_pending( + connection, + within=frozenset(path.resolve().relative_to(package).as_posix() for path in owned), + ) + for path in owned: await connection.execute(path.read_text()) if self._change_feed: await connection.execute(_CHANGES_DDL_PATH.read_text()) 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 index 62fd77ab0..f5bbeca4a 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py @@ -18,9 +18,11 @@ Every migration runs in its own transaction under the reporting schema lock, so concurrent runners serialize and a failure commits nothing. -``PgReportingLedgerStore.create_schema()`` applies the same files, recorded the -same way, inside its own bootstrap transaction (:func:`apply_pending`), so a -ledger created or upgraded by either entry point carries its schema version. +``PgReportingLedgerStore.create_schema()`` applies the migrations made of the +DDL it owns (``0001``–``0003``), recorded the same way, inside its own +bootstrap transaction (:func:`apply_pending`), so a ledger created or upgraded +by either entry point carries its schema version; ``python -m adcp migrate`` +then adds what only ``postgres_storage`` needs (``0004``, ``0005``). """ from __future__ import annotations @@ -194,13 +196,20 @@ async def _migrate(connection: Any, *, dry_run: bool) -> list[Migration]: return applied -async def apply_pending(connection: Any) -> list[Migration]: - """Apply every pending migration on ``connection`` inside the caller's transaction. +async def apply_pending( + connection: Any, *, within: frozenset[str] | None = None +) -> list[Migration]: + """Apply pending migrations on ``connection`` inside the caller's transaction. The bootstrap path of ``create_schema()``: the caller holds one transaction for the whole upgrade, so every migration commits with it or none does. Takes the reporting schema lock first, like :func:`migrate`, so a booting worker and ``python -m adcp migrate`` serialize instead of racing. + + ``within`` bounds what the caller owns: migrations are applied in order + and the walk stops at the first pending one that includes a file outside + it (a later migration may depend on an earlier one, so none is skipped). + The rest stay pending for ``python -m adcp migrate``. """ await connection.execute(_LOCK) await connection.execute(_TABLE) @@ -209,6 +218,8 @@ async def apply_pending(connection: Any) -> list[Migration]: for migration in migrations(): if migration.version in applied: continue + if within is not None and not set(migration.includes) <= within: + break if await _apply(connection, migration): result.append(migration) return result diff --git a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py index 522baed80..33e94cd6f 100644 --- a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py @@ -12,8 +12,9 @@ * no retained evidence row changes beyond the added ``offering_id = ''``; * every record reads back and a restart rebinds the stored generations; * the upgraded catalog is identical to a freshly migrated one; -* the applied schema version is recorded, re-running is a no-op, and - concurrent boots serialize; +* the applied schema versions are recorded (``create_schema()`` records the + ledger's own ``0001``–``0003``; ``migrate`` the rest), re-running is a + no-op, and concurrent boots serialize; * ``adcp==10.0.0b4`` with *this* storage, the pair a seller upgrades to, upgrades a 0.2.0b0 ledger, where the published 0.2.0b0 fails. @@ -246,6 +247,7 @@ async def test_create_schema_upgrades_a_published_ledger_in_place( ) -> None: """The ``store=`` startup path (``ReliableReportingService`` calls ``create_schema``).""" from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from adcp_reporting_storage.migrations import migrate from adcp.storage.postgres import postgres_storage @@ -258,15 +260,25 @@ async def test_create_schema_upgrades_a_published_ledger_in_place( store = PgReportingLedgerStore(pool=pool) await store.create_schema() - assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + # The ledger's own migrations are applied and recorded; the ones that + # only postgres_storage needs stay pending for python -m adcp migrate. + assert await _versions(pool) == ["0001", "0002", "0003"] after = await _image_here(pool) assert_evidence_unchanged(seeded["image"], after) assert await _readback_here(pool) == seeded["readback"] - await postgres_storage(pool=pool).check_schema() # the migrations-managed path agrees + upgraded = await _catalog(pool) await store.create_schema() # re-runnable: nothing changes assert await _image_here(pool) == after - assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + assert await _versions(pool) == ["0001", "0002", "0003"] + + applied = await migrate(pool) + assert [item.version for item in applied] == ["0004", "0005"] + await postgres_storage(pool=pool).check_schema() # the migrations-managed path agrees + assert_evidence_unchanged(seeded["image"], await _image_here(pool)) + async with isolated_reporting_pool(autocommit=True) as fresh: + await PgReportingLedgerStore(pool=fresh).create_schema() + assert upgraded == await _catalog(fresh) async def test_migrate_upgrades_a_published_ledger_to_the_fresh_catalog( @@ -382,7 +394,7 @@ async def test_concurrent_boots_and_migrate_serialize_on_a_published_ledger( row = await ( await connection.execute("SELECT count(*) FROM adcp_schema_migrations") ).fetchone() - assert row[0] == 5 + assert row[0] == 5 # each version recorded once across the four runners assert_evidence_unchanged(seeded["image"], await _image_here(pool)) @@ -408,4 +420,4 @@ async def test_adcp_b4_fails_on_the_published_storage_and_upgrades_with_this_one assert "error" not in upgraded, upgraded["error"] assert upgraded["readback"] == seeded["readback"] assert_evidence_unchanged(seeded["image"], upgraded["image"]) - assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + assert await _versions(pool) == ["0001", "0002", "0003"] diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 6c21c58f2..4b2b9e260 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -9045,6 +9045,7 @@ "ReportingRowQuarantineReason": "union[value:'BINDING_MISMATCH',value:'ROWS_NOT_PORTABLE',value:'ROW_SEQUENCE_INVALID']", "ReportingRowStorageOptionsV1": "adcp_reporting_storage.ledger.row_storage", "ReportingRowStorageV1": "adcp_reporting_storage.ledger.row_storage", + "STORAGE_CONTRACT": "value:2", "assert_gcs_reporting_row_bucket_policy_v1": "adcp_reporting_storage.ledger.row_storage_gcs", "backfill_legacy_reporting": "adcp_reporting_storage.migration", "create_azure_blob_reporting_row_object_provider_v1": "adcp_reporting_storage.ledger.row_storage_azure", @@ -9060,6 +9061,7 @@ "replace_empty_owned_reporting": "adcp_reporting_storage.migration", "replace_empty_owned_reporting_sync": "adcp_reporting_storage.migration", "reporting_period_date_v1": "adcp_reporting_storage.ledger.row_storage_object", + "require_adcp_contract": "adcp_reporting_storage.contract", "require_owned_schema_or_empty": "adcp_reporting_storage.migration", "run_reporting_row_object_provider_conformance_v1": "adcp_reporting_storage.ledger.row_storage_object" } diff --git a/tests/test_release_configuration.py b/tests/test_release_configuration.py index c8621978c..9a6eaf4ad 100644 --- a/tests/test_release_configuration.py +++ b/tests/test_release_configuration.py @@ -150,15 +150,18 @@ def test_release_please_keeps_next_on_the_beta_channel() -> None: def test_storage_distribution_pins_one_adcp_major() -> None: - """The storage distribution's adcp range is one major on main. + """The storage distribution's adcp range is one major, floored at its contract. - On the next draft-spec line the range also spans adcp 10 pre-releases, or - next's first ``10.0.0aN`` release could not install the storage package - from the workspace. next never releases the storage distribution; its - first adcp-10 release sets the one-major range again (see RELEASING.md). + The floor is the first adcp release of the storage contract the package + implements (``adcp_reporting_storage.contract``): on next that is + 10.0.0b4, which added ``offering_id`` to the ledger column lists; a wider + floor let 10.0.0b4 install beside a storage release that lacked the + column (#1763). The ceiling stays the next major (see RELEASING.md). """ pyproject = (STORAGE / "pyproject.toml").read_text() if (ROOT / "release-please-config.next.json").exists(): - assert re.search(r'"adcp>=9\.0\.0(?:b\d+)?,<11"', pyproject), "next spans adcp 9 and 10" + assert re.search( + r'"adcp>=10\.0\.0b4,<11"', pyproject + ), "next's storage floor is the first adcp of its storage contract" else: assert re.search(r'"adcp>=9\.0\.0(?:b\d+)?,<10"', pyproject), "adcp range must be one major" From c596855f00c00aac4b39bd6c9d64fd5358fea5ce Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 04:56:12 -0500 Subject: [PATCH 3/5] fix(reporting): refuse an adcp-reporting-storage of another storage contract at startup (#1789) adcp 10.0.0b4 added offering_id to the ledger column lists that adcp-reporting-storage reads against its DDL; the published storage release (0.2.0b0) lacked the column, installed cleanly beside 10.0.0b4, and failed every reporting turn with UndefinedColumn, logged only as a configuration failure (#1763). adcp now names the ledger layout it reads with one integer, adcp.reporting.ledger.storage_contract.REPORTING_STORAGE_CONTRACT (2), and checks the installed ledger store against it at service startup (ReliableReportingService._initialize, before any schema work) and when postgres_storage() is built. A store declares its contract with a storage_contract attribute (adcp-reporting-storage 0.2.0b1+, #1787); a store from that distribution without one predates the declaration and is contract 1; other stores are not subject to the check. A mismatch raises ReliableReportingConfigurationError with REPORTING_SCHEMA_BEHIND, naming both installed versions, both contracts, the release to install and the migration to run, so the pair fails once and loudly instead of every turn. tests/test_reporting_storage_contract.py pins a fingerprint of the shared column lists to the contract number, so changing _record_rows without bumping the contract (and the storage DDL with it) fails the build. Co-authored-by: Claude Fable 5.1 --- MIGRATION_v9_to_v10.md | 21 +++ docs/reliable-reporting-service.md | 4 + docs/reporting-subscriptions-errors.md | 2 +- docs/reporting-subscriptions-seller.md | 2 +- src/adcp/reporting/ledger/storage_contract.py | 124 ++++++++++++++++ src/adcp/reporting/service.py | 5 + src/adcp/storage/__init__.py | 3 + tests/test_reporting_storage_contract.py | 136 ++++++++++++++++++ 8 files changed, 295 insertions(+), 2 deletions(-) create mode 100644 src/adcp/reporting/ledger/storage_contract.py create mode 100644 tests/test_reporting_storage_contract.py diff --git a/MIGRATION_v9_to_v10.md b/MIGRATION_v9_to_v10.md index cd705b1e5..c314be92b 100644 --- a/MIGRATION_v9_to_v10.md +++ b/MIGRATION_v9_to_v10.md @@ -53,3 +53,24 @@ the SDK-relevant inventory is [docs/adcp-3.3-sdk-inventory.md](docs/adcp-3.3-sdk | Generated enums name each value once | The `_1` alias members codegen emitted when an `anyOf` arm repeated an enum value (`System.zip_1`, `Field17.ad_request_url_1`, 19 enums on 10.0.0b4) are gone; each value has one member, the first-declared one. Code that referenced an alias name (`System.zip_1`) must use the real member (`System.zip`); `System("zip")` and `member.value` are unchanged. | | `list_account_changes` pages are checked and projected per caller | AdCP 3.3 requires non-empty `changed_paths` on every `adcp` / `seller_operator` change record (`[""]` for a whole-resource create, delete or purge) and lets the originating request's `idempotency_key` reach only the principal that sent it. A handler on 3.3 or later that returns such a record without `changed_paths` now fails the call with `VALIDATION_ERROR` (`field` `/changes/N/changed_paths`), whatever the validation mode; and every record loses `idempotency_key` unless it is an `adcp` record built with `adcp.account_change_feed.attribute_account_change(record, originator)` whose originator is the caller (`ToolContext.tenant_id` + `caller_identity`). Persist the originator when the mutation commits and attribute records when serving the page. `WebhookSender.prepare_account_change_recorded` / `send_account_change_recorded` also accept the change record itself (with `subscriber_id=`), refuse one without `changed_paths`, and never carry its `idempotency_key`. | | `get_adcp_capabilities` reports `adcp.implementation` | `serve()` (and `adcp.decisioning.serve()`) now adds AdCP 3.3 `adcp.implementation.sdks[]` to capabilities served at 3.3 or later: one entry `{"name": "adcp", "version": , "components": [...]}`, listing only components the server provably wires agent-wide (`transport` always; `types` for Decisioning platforms; `schema_validation` with strict request and response validation and no `2.5` in `supported_versions`; `authentication` with `auth=BearerTokenAuth(...)` and no widened discovery set; `idempotency` when every advertised mutating task is `@IdempotencyStore.wrap`ped; `webhooks` / `signing` only when the SDK owns and signs every outbound webhook; `lifecycle` never automatically). A 3.0 to 3.2 caller never sees it. Tests that compare a whole 3.3 capabilities response need the new key. Opt out with `serve(report_implementation=False)` (or `ServeConfig(report_implementation=False)`); add components or other SDKs with `ImplementationReport(extra_components=[...], extra_sdks=[...])`; or declare the block yourself (`Adcp(implementation=...)` on `DecisioningCapabilities`, `capabilities_response(implementation=...)`, or in your own handler's response), which the SDK never overwrites. `adcp.server.adcp_sdk_entry()` builds this SDK's entry for a hand-written block. `DecisioningCapabilities.adcp.implementation` was previously dropped from the projection; it is now emitted. | + +## Reporting storage contract (adcp-reporting-storage) + +adcp 10.0.0b4 widened the reporting ledger's generation key with `offering_id` +(AdCP 3.3 reporting subscriptions). The layout `adcp` and +`adcp-reporting-storage` share is now named by one integer on each side, +`adcp.reporting.ledger.storage_contract.REPORTING_STORAGE_CONTRACT` and +`adcp_reporting_storage.STORAGE_CONTRACT` (2 for SDK 10; SDK 9 and +adcp 10.0.0b0–b3 are 1): + +- `adcp-reporting-storage` 0.2.0b1 requires `adcp>=10.0.0b4,<11`; the + 0.2.0b0 / 0.1.0b4 releases do not work with 10.0.0b4 or later + ([#1763](https://github.com/adcontextprotocol/adcp-client-python/issues/1763)). +- Startup (`ReliableReportingService`, `postgres_storage`) refuses a pair whose + contracts differ with `REPORTING_SCHEMA_BEHIND`, naming both versions and + the release to install. +- An existing SDK 9 ledger is upgraded in place by `create_schema()` or + `python -m adcp migrate`: `offering_id` is added with `''` for every + existing account-level generation, obligation and status, and the applied + schema versions are recorded. Stop every SDK 9 reporting worker first; see + [docs/reporting-ledger-migration.md](docs/reporting-ledger-migration.md). diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 7353bfc3a..d4801e988 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -412,6 +412,10 @@ platform = reporting.install(MySeller()) # then serve the platform as usual (`--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. + Startup also refuses an installed `adcp-reporting-storage` whose storage + contract differs from this adcp's (`adcp.reporting.ledger.storage_contract`), + with the same code, naming both versions and the release to install, so an + incompatible pair fails once at startup rather than on every turn (#1763). - `0002 reporting_subscription_generations` widens the generation key and `0006 reporting_status_checkpoint_offerings` keys status checkpoints and status events by the generation's resolved offering. Both are diff --git a/docs/reporting-subscriptions-errors.md b/docs/reporting-subscriptions-errors.md index bc512a77d..597ba42ff 100644 --- a/docs/reporting-subscriptions-errors.md +++ b/docs/reporting-subscriptions-errors.md @@ -16,7 +16,7 @@ Every SDK error carries `code` and `recovery` (`transient`, `correctable` or `te | `COVERAGE_INPUT_INCOMPLETE`, `COVERAGE_INPUT_INVALID` | `CoverageInputError` | n/a | correctable | supply the named input | | `COVERAGE_BUDGET_EXCEEDED` | `CoverageInputError` | n/a | correctable | narrow the input | | `REPORTING_STORAGE_NOT_DURABLE` | `ReliableReportingConfigurationError` | n/a | correctable | `storage=postgres_storage(...)`, or `ADCP_ENV=development` | -| `REPORTING_SCHEMA_BEHIND` | `ReliableReportingConfigurationError` | n/a | correctable | run `python -m adcp migrate` | +| `REPORTING_SCHEMA_BEHIND` | `ReliableReportingConfigurationError` | n/a | correctable | run `python -m adcp migrate`; or, when the installed `adcp-reporting-storage` implements another storage contract than this `adcp` ([#1763](https://github.com/adcontextprotocol/adcp-client-python/issues/1763)), install the release the message names | | `SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS` | `ReliableReportingConfigurationError` | n/a | correctable | publish `standard_dimensions` on the named offering | | `SUBSCRIPTION_ROSTER_UNAVAILABLE` | `ReliableReportingConfigurationError` | n/a | correctable | serve `list_accounts`, or supply `SubscriptionOptions.accounts` | diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index 3920ddc82..1b9d40be4 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -103,7 +103,7 @@ account.authorization = wind_down_grant(account.authorization) Pass one storage object: `memory_storage()` in development, `postgres_storage(dsn=... | pool=...)` in production. Process-local storage is refused outside development (`REPORTING_STORAGE_NOT_DURABLE`). -Apply schema changes with `python -m adcp migrate [--status | --dry-run]` before starting a release. Startup refuses a schema that is behind (`REPORTING_SCHEMA_BEHIND`). `0002 reporting_subscription_generations` and `0006 reporting_status_checkpoint_offerings` are stopped-worker upgrades: stop every reporting worker first. +Apply schema changes with `python -m adcp migrate [--status | --dry-run]` before starting a release. Startup refuses a schema that is behind (`REPORTING_SCHEMA_BEHIND`), and an installed `adcp-reporting-storage` of another storage contract than this `adcp` (same code; the message names the release to install). `0002 reporting_subscription_generations` and `0006 reporting_status_checkpoint_offerings` are stopped-worker upgrades: stop every reporting worker first. ## Workers and scaling diff --git a/src/adcp/reporting/ledger/storage_contract.py b/src/adcp/reporting/ledger/storage_contract.py new file mode 100644 index 000000000..4818280ce --- /dev/null +++ b/src/adcp/reporting/ledger/storage_contract.py @@ -0,0 +1,124 @@ +"""The reporting storage contract: the ledger layout adcp and adcp-reporting-storage share. + +adcp reads and writes the PostgreSQL ledger through the column lists in +:mod:`adcp.reporting.ledger._record_rows`; ``adcp-reporting-storage`` ships +the DDL those lists are read against. The two must agree exactly: a column one +side knows and the other does not fails every reporting turn with +``UndefinedColumn``, logged only as a configuration failure +(adcp-client-python#1763). The agreement is named by one integer on each side, +bumped together whenever the shared layout changes: + +* **1** — the SDK 9 layout and adcp 10.0.0b0–b3: a configuration generation is + ``(account_id, consumer_id, delivery_config_id, delivery_config_version)``. + Implemented by adcp-reporting-storage 0.1.0b4 and 0.2.0b0. +* **2** — adcp 10.0.0b4 and later: ``offering_id`` on configurations, + obligations and consumer statuses (AdCP 3.3 reporting subscriptions). + Implemented by adcp-reporting-storage 0.2.0b1 and later. + +:func:`require_storage_contract` runs at service startup and when +``postgres_storage`` is built, so an incompatible pair is refused once, with +both versions named and the fix spelled out, instead of failing every turn. +``tests/test_reporting_storage_contract.py`` pins the column lists to the +contract number, so changing them without bumping it fails the build; the +companion release gate (``scripts/check_companion_releases.py``) refuses to +publish either package against a companion of another contract. +""" + +from __future__ import annotations + +import hashlib +from importlib.metadata import PackageNotFoundError, version +from typing import Any + +__all__ = [ + "REPORTING_STORAGE_CONTRACT", + "declared_storage_contract", + "require_storage_contract", + "storage_contract_fingerprint", +] + +#: The storage contract this adcp reads the reporting ledger with. +REPORTING_STORAGE_CONTRACT = 2 + +#: The distribution that implements the contract for PostgreSQL. +STORAGE_DISTRIBUTION = "adcp-reporting-storage" +#: Its first release of the current contract. +STORAGE_RELEASE = "0.2.0b1" + + +def storage_contract_fingerprint() -> str: + """SHA-256 over the shared column lists; pinned per contract number in the tests.""" + from adcp.reporting.ledger import _record_rows + + lists = ( + _record_rows._CONFIGURATION_COLUMNS, + _record_rows._OBLIGATION_COLUMNS, + _record_rows._REVISION_COLUMNS, + _record_rows._ADJUSTMENT_COLUMNS, + _record_rows._STATUS_COLUMNS_BARE, + _record_rows._ISSUE_COLUMNS, + ) + normalized = "\n".join( + ",".join(column.strip() for column in columns.split(",")) for columns in lists + ) + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def declared_storage_contract(store: Any) -> int | None: + """The contract ``store`` implements, or ``None`` for a store that declares none. + + A store declares it with a ``storage_contract`` attribute. A store from + the ``adcp_reporting_storage`` distribution without one predates the + declaration (0.1.0b4, 0.2.0b0) and implements contract 1. Any other + store (in-memory, an adopter's own) is not subject to the check. + """ + declared = getattr(store, "storage_contract", None) + if declared is not None: + return int(declared) + module = getattr(type(store), "__module__", "") or "" + if module == "adcp_reporting_storage" or module.startswith("adcp_reporting_storage."): + return 1 + return None + + +def _installed(distribution: str) -> str: + try: + return version(distribution) + except PackageNotFoundError: + return "unknown" + + +def require_storage_contract(store: Any) -> None: + """Refuse a ledger store whose storage contract differs from this adcp's. + + Raises :class:`adcp.reporting.ReliableReportingConfigurationError` with + code ``REPORTING_SCHEMA_BEHIND`` (recovery ``correctable``), naming both + installed versions, both contracts, the release to install and the + migration to run. + """ + implemented = declared_storage_contract(store) + if implemented is None or implemented == REPORTING_STORAGE_CONTRACT: + return + from adcp.reporting._errors import ReliableReportingConfigurationError + + adcp_version = _installed("adcp") + storage_version = _installed(STORAGE_DISTRIBUTION) + if implemented < REPORTING_STORAGE_CONTRACT: + fix = ( + f"Install {STORAGE_DISTRIBUTION}>={STORAGE_RELEASE} (storage contract " + f'{REPORTING_STORAGE_CONTRACT}), then run "python -m adcp migrate" or start the ' + "service, whose create_schema() upgrades the ledger in place." + ) + else: + fix = ( + f"Install the adcp release that reads storage contract {implemented}, or " + f"{STORAGE_DISTRIBUTION} of contract {REPORTING_STORAGE_CONTRACT}." + ) + raise ReliableReportingConfigurationError( + f"adcp {adcp_version} reads the reporting ledger with storage contract " + f"{REPORTING_STORAGE_CONTRACT}, but the installed {STORAGE_DISTRIBUTION} " + f"{storage_version} implements contract {implemented}; the pair would fail every " + f"reporting turn. {fix}", + code="REPORTING_SCHEMA_BEHIND", + config_field=STORAGE_DISTRIBUTION, + ) diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 0389e6e7b..e613f2282 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -1426,6 +1426,11 @@ def _require_durable_subscriptions(self) -> None: async def _initialize(self) -> None: self.validate() self._require_durable_subscriptions() + from adcp.reporting.ledger.storage_contract import require_storage_contract + + # A storage distribution of another contract fails once here, naming + # the pair and the fix, never on every turn (#1763). + require_storage_contract(self.store) async with self._configuration_lock: 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. diff --git a/src/adcp/storage/__init__.py b/src/adcp/storage/__init__.py index 5701e803e..ebbe7601d 100644 --- a/src/adcp/storage/__init__.py +++ b/src/adcp/storage/__init__.py @@ -102,10 +102,13 @@ def postgres_storage( Resolved through the distribution's ``adcp.storage`` entry point, so ``adcp`` never imports it by name. """ + from adcp.reporting.ledger.storage_contract import require_storage_contract + 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") + require_storage_contract(storage.ledger) # an incompatible distribution fails here return storage diff --git a/tests/test_reporting_storage_contract.py b/tests/test_reporting_storage_contract.py new file mode 100644 index 000000000..07a57017e --- /dev/null +++ b/tests/test_reporting_storage_contract.py @@ -0,0 +1,136 @@ +"""adcp refuses an adcp-reporting-storage of another storage contract at startup (#1763).""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from adcp.reporting import ReliableReportingConfigurationError +from adcp.reporting.ledger import InMemoryReportingLedgerStore +from adcp.reporting.ledger import storage_contract as contract +from adcp.reporting.ledger.storage_contract import ( + REPORTING_STORAGE_CONTRACT, + declared_storage_contract, + require_storage_contract, + storage_contract_fingerprint, +) + +#: The shared column lists, pinned per contract. Changing a list in +#: ``_record_rows`` changes the fingerprint: bump REPORTING_STORAGE_CONTRACT, +#: ship the matching adcp-reporting-storage DDL and migration, and re-pin. +FINGERPRINTS = {2: "e6cb6454085a8fdc004f564f3882bcfe1a67fd9e622d72a36ba1f9d6ccd12af1"} + + +def test_the_column_lists_are_pinned_to_the_contract_number() -> None: + assert REPORTING_STORAGE_CONTRACT in FINGERPRINTS, "pin the new contract's fingerprint" + assert storage_contract_fingerprint() == FINGERPRINTS[REPORTING_STORAGE_CONTRACT], ( + "the shared ledger column lists changed: bump REPORTING_STORAGE_CONTRACT together " + "with adcp-reporting-storage's STORAGE_CONTRACT, DDL and migration, then re-pin" + ) + + +class _Declared: + storage_contract = REPORTING_STORAGE_CONTRACT + + +class _Older: + storage_contract = REPORTING_STORAGE_CONTRACT - 1 + + +class _Newer: + storage_contract = REPORTING_STORAGE_CONTRACT + 1 + + +class _Undeclared: + """A store from adcp-reporting-storage 0.1.0b4/0.2.0b0: no declaration at all.""" + + +_Undeclared.__module__ = "adcp_reporting_storage.ledger.pg" + + +class _Custom: + """An adopter's own store, outside the check.""" + + +@pytest.mark.parametrize( + ("store", "expected"), + [(_Declared(), 2), (_Older(), 1), (_Newer(), 3), (_Undeclared(), 1), (_Custom(), None)], + ids=["declared", "older", "newer", "undeclared-storage-release", "custom"], +) +def test_a_store_is_classified_by_declaration_or_provenance( + store: Any, expected: int | None +) -> None: + assert declared_storage_contract(store) == expected + + +def test_the_in_memory_store_and_custom_stores_pass() -> None: + require_storage_contract(InMemoryReportingLedgerStore()) + require_storage_contract(_Custom()) + require_storage_contract(_Declared()) + + +def test_a_storage_release_of_an_older_contract_is_refused_with_the_fix() -> None: + with pytest.raises(ReliableReportingConfigurationError) as refused: + require_storage_contract(_Undeclared()) + error = refused.value + assert error.code == "REPORTING_SCHEMA_BEHIND" and error.recovery == "correctable" + assert error.config_field == "adcp-reporting-storage" + message = str(error) + assert f"storage contract {REPORTING_STORAGE_CONTRACT}" in message + assert "implements contract 1" in message + assert "Install adcp-reporting-storage>=0.2.0b1" in message + assert 'run "python -m adcp migrate"' in message and "create_schema()" in message + + +def test_a_storage_release_of_a_newer_contract_is_refused_too() -> None: + with pytest.raises(ReliableReportingConfigurationError, match="implements contract 3"): + require_storage_contract(_Newer()) + + +async def test_service_startup_refuses_the_pair_before_touching_the_schema( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from adcp.reporting.service import ReliableReportingService + from tests.test_reliable_reporting_service import NOW, _account_context + + store = InMemoryReportingLedgerStore(clock=lambda: NOW) + monkeypatch.setattr(InMemoryReportingLedgerStore, "storage_contract", 1, raising=False) + touched = [] + + async def create_schema() -> None: + touched.append(True) + + monkeypatch.setattr(store, "create_schema", create_schema) + service = ReliableReportingService( + store=store, clock=lambda: NOW, account_context=_account_context + ) + with pytest.raises(ReliableReportingConfigurationError) as refused: + await service.initialize() + assert refused.value.code == "REPORTING_SCHEMA_BEHIND" + assert touched == [], "the schema must not be created or upgraded on a refused pair" + + +def test_postgres_storage_refuses_the_pair_at_construction(monkeypatch: pytest.MonkeyPatch) -> None: + import adcp.storage as facade + from adcp.storage import ReportingStorage + + def build(**kwargs: Any) -> ReportingStorage: + return ReportingStorage( + durability="durable", ledger=_Undeclared(), principals=object(), subscriptions=object() + ) + + monkeypatch.setattr(facade, "_plugin", lambda group, name: build) + with pytest.raises(ReliableReportingConfigurationError, match="implements contract 1"): + facade.postgres_storage(dsn="postgresql://unused") + + +def test_the_installed_storage_distribution_declares_this_contract() -> None: + pytest.importorskip("adcp_reporting_storage") + from adcp_reporting_storage.contract import STORAGE_CONTRACT, adcp_storage_contract + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + assert STORAGE_CONTRACT == REPORTING_STORAGE_CONTRACT + assert adcp_storage_contract() == REPORTING_STORAGE_CONTRACT # storage reads our declaration + assert declared_storage_contract.__module__ == contract.__name__ + assert PgReportingLedgerStore.storage_contract == REPORTING_STORAGE_CONTRACT From 15226447dce538e1bd28f134a07d84f739950bba Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 10:14:22 +0000 Subject: [PATCH 4/5] test(reporting-storage): follow the shipped migration set and cover a notifications-enabled ledger next added migration 0006 (status checkpoint offerings) after this branch was cut; the published-wheel upgrade test now derives the versions it expects from the shipped set instead of listing them, and seeds a notifications-enabled 0.2.0b0 ledger (outbox and status stores) to prove the in-place upgrade and migrate record every migration, 0006 included, with the evidence unchanged. Co-Authored-By: Claude Fable 5.1 --- docs/reporting-ledger-migration.md | 3 +- packages/adcp-reporting-storage/README.md | 3 +- .../src/adcp_reporting_storage/ledger/pg.py | 9 ++- .../migrations/__init__.py | 3 +- .../reporting/_published_ledger_seed.py | 22 +++++- ..._reporting_storage_published_upgrade_pg.py | 76 +++++++++++++++---- 6 files changed, 93 insertions(+), 23 deletions(-) diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md index 765a9a6d2..6a98f6ec3 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -304,7 +304,8 @@ primary key and the obligation period index with the new column; and re-points the managed-delivery foreign keys. No evidence row is rewritten or dropped. The applied versions are recorded in `adcp_schema_migrations` (`create_schema()` applies the ledger's own `0001`–`0003`; `python -m adcp migrate` also applies -`0004`–`0005`, which only `postgres_storage` needs), re-running is a no-op, +the later ones, which only `postgres_storage` and the notification stores +need), re-running is a no-op, and concurrent boots serialize on the reporting schema advisory lock inside one transaction. It is a stopped-worker upgrade: stop every worker of the earlier release first, because their configuration writes name the diff --git a/packages/adcp-reporting-storage/README.md b/packages/adcp-reporting-storage/README.md index dd28ad4d3..63d542e15 100644 --- a/packages/adcp-reporting-storage/README.md +++ b/packages/adcp-reporting-storage/README.md @@ -80,7 +80,8 @@ tables, backfilled with `''` for every existing account-level generation, the generation keys and unique indexes are widened, and the applied schema versions are recorded in `adcp_schema_migrations` (`create_schema()` applies and records the ledger's own `0001`–`0003`; `python -m adcp migrate` also -applies `0004`–`0005`, which only `postgres_storage` needs). No evidence row is +applies the later ones, which only `postgres_storage` and the notification +stores need). No evidence row is rewritten or dropped. The upgrade is transactional, serialized under the reporting schema advisory lock, and a no-op when re-run. It is a **stopped-worker** upgrade: stop every reporting worker of the earlier release first (its configuration diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 119cc0011..53cc6bbe6 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -486,8 +486,9 @@ async def create_schema(self) -> None: 0.1.0b4 or 0.2.0b0 is upgraded in place: ``offering_id`` is added with ``''`` for every existing account-level generation, and the applied versions are recorded, so ``python -m adcp migrate --status`` reads - them. Re-running is a no-op. ``postgres_storage`` additionally needs - ``0004``/``0005``, which only ``python -m adcp migrate`` applies. + them. Re-running is a no-op. The later migrations (``0004`` onward: subscription + state, inline storage, status checkpoint offerings) are applied and + recorded by ``python -m adcp migrate``, or by the stores that own them. The bootstrap takes a transaction-scoped schema lock before any DDL. Keep the migration in that transaction, including with an autocommit @@ -700,8 +701,8 @@ async def _create_schema_on(self, connection: Any) -> None: # recorded as applied, so a 0.1.0b4/0.2.0b0 ledger is upgraded in place # and carries its version; the idempotent files below then only # re-verify what they installed. Migrations of other stores' tables - # (0004 subscription state, 0005 inline storage) stay pending for - # python -m adcp migrate: the catalog contract of this store's older + # (0004 subscription state, 0005 inline storage, 0006 status checkpoint + # offerings) stay pending for python -m adcp migrate: the catalog contract of this store's older # binaries is exactly its own reporting_* objects. package = Path(__file__).resolve().parents[1] await apply_pending( 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 index f5bbeca4a..855b4e96a 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py @@ -22,7 +22,8 @@ DDL it owns (``0001``–``0003``), recorded the same way, inside its own bootstrap transaction (:func:`apply_pending`), so a ledger created or upgraded by either entry point carries its schema version; ``python -m adcp migrate`` -then adds what only ``postgres_storage`` needs (``0004``, ``0005``). +then adds the later ones (``0004`` onward), which only ``postgres_storage`` +and the notification stores need. """ from __future__ import annotations diff --git a/tests/conformance/reporting/_published_ledger_seed.py b/tests/conformance/reporting/_published_ledger_seed.py index 3b2c1912a..964d61a8b 100644 --- a/tests/conformance/reporting/_published_ledger_seed.py +++ b/tests/conformance/reporting/_published_ledger_seed.py @@ -4,7 +4,9 @@ creates the ledger with the installed pair's own ``create_schema()`` and writes two account-level generations with obligations, revisions and rows, the way a seller on that release did; ``read`` reads them back; ``upgrade`` runs the -installed storage's ``create_schema()`` first. Every action returns an ``image`` +installed storage's ``create_schema()`` first. With ``notifications`` the +outbox and status-notification stores are created too, as a seller with +status notifications boots them. Every action returns an ``image`` of every ``reporting_*``/``adcp_reporting_*`` table so the test can prove no retained evidence changed. Only APIs common to adcp 9.3.1 and 10.0.0b2+ are used. """ @@ -172,19 +174,33 @@ async def readback(store): return result +async def create_schema(pool, store, *, notifications): + """The stores a notifications-enabled seller boots, in the order it boots them.""" + if notifications: + from adcp_reporting_storage import PgReportingOutbox + + await PgReportingOutbox(pool=pool).create_schema() + await store.create_schema() + if notifications: + from adcp_reporting_storage.outbox.status_pg import PgStatusNotificationStore + + await PgStatusNotificationStore(store).create_schema() + + async def main(settings): from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore from psycopg_pool import AsyncConnectionPool + notifications = bool(settings.get("notifications")) async with AsyncConnectionPool( settings["conninfo"], kwargs=settings["kwargs"], min_size=1, max_size=4, open=False ) as pool: await pool.open() - store = PgReportingLedgerStore(pool=pool) + store = PgReportingLedgerStore(pool=pool, notifications=notifications) result = {"action": settings["action"]} try: if settings["action"] in {"seed", "upgrade"}: - await store.create_schema() + await create_schema(pool, store, notifications=notifications) if settings["action"] == "seed": for account_id in ACCOUNTS: configuration = _configuration(account_id) diff --git a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py index 33e94cd6f..aab61c5ac 100644 --- a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py @@ -61,6 +61,9 @@ "reporting_materializer_discovery", "reporting_status_scope_checkpoints", "reporting_production_generations", + # Migration 0006 (status checkpoint offerings) re-keys these the same way. + "reporting_status_checkpoints", + "reporting_status_events", } ACCOUNTS = ("account-redacted", "account-b") SEEDED_KEYS = { @@ -174,12 +177,21 @@ def b4_with_this_storage(workspace: Path) -> Path: return python -def _settings(pool: Any, action: str) -> dict[str, Any]: - return {"conninfo": pool.conninfo, "kwargs": pool.kwargs, "action": action} +def _settings(pool: Any, action: str, *, notifications: bool = False) -> dict[str, Any]: + return { + "conninfo": pool.conninfo, + "kwargs": pool.kwargs, + "action": action, + "notifications": notifications, + } -async def _seed(python: Path, pool: Any, cwd: Path) -> dict[str, Any]: - result = await asyncio.to_thread(_run, python, _settings(pool, "seed"), cwd=cwd) +async def _seed( + python: Path, pool: Any, cwd: Path, *, notifications: bool = False +) -> dict[str, Any]: + result = await asyncio.to_thread( + _run, python, _settings(pool, "seed", notifications=notifications), cwd=cwd + ) assert "error" not in result, result["error"] assert sorted(result["readback"]["revisions"]) == sorted( f"rpr_{account}_{ordinal}" for account in ACCOUNTS for ordinal in range(2) @@ -209,11 +221,25 @@ def assert_evidence_unchanged(before: dict[str, Any], after: dict[str, Any]) -> assert after[table] == [], f"the upgrade wrote rows into new table {table}" +#: The migrations create_schema() owns and records; the rest are migrate's. +LEDGER_VERSIONS = ["0001", "0002", "0003"] + + +def _shipped() -> list[str]: + from adcp_reporting_storage.migrations import migrations + + return [item.version for item in migrations()] + + +def _rest() -> list[str]: + return [version for version in _shipped() if version not in LEDGER_VERSIONS] + + async def _versions(pool: Any) -> list[str]: - from adcp_reporting_storage.migrations import migration_status, migrations + from adcp_reporting_storage.migrations import migration_status statuses = await migration_status(pool) - assert [item.migration.version for item in statuses] == [m.version for m in migrations()] + assert [item.migration.version for item in statuses] == _shipped() return [item.migration.version for item in statuses if not item.pending] @@ -262,7 +288,7 @@ async def test_create_schema_upgrades_a_published_ledger_in_place( # The ledger's own migrations are applied and recorded; the ones that # only postgres_storage needs stay pending for python -m adcp migrate. - assert await _versions(pool) == ["0001", "0002", "0003"] + assert await _versions(pool) == LEDGER_VERSIONS after = await _image_here(pool) assert_evidence_unchanged(seeded["image"], after) assert await _readback_here(pool) == seeded["readback"] @@ -270,10 +296,10 @@ async def test_create_schema_upgrades_a_published_ledger_in_place( await store.create_schema() # re-runnable: nothing changes assert await _image_here(pool) == after - assert await _versions(pool) == ["0001", "0002", "0003"] + assert await _versions(pool) == LEDGER_VERSIONS applied = await migrate(pool) - assert [item.version for item in applied] == ["0004", "0005"] + assert [item.version for item in applied] == _rest() await postgres_storage(pool=pool).check_schema() # the migrations-managed path agrees assert_evidence_unchanged(seeded["image"], await _image_here(pool)) async with isolated_reporting_pool(autocommit=True) as fresh: @@ -291,7 +317,7 @@ async def test_migrate_upgrades_a_published_ledger_to_the_fresh_catalog( async with isolated_reporting_pool(autocommit=True) as pool: seeded = await _seed(python, pool, workspace) applied = await migrate(pool) - assert [item.version for item in applied] == ["0001", "0002", "0003", "0004", "0005"] + assert [item.version for item in applied] == _shipped() assert await migrate(pool) == [] assert_evidence_unchanged(seeded["image"], await _image_here(pool)) assert await _readback_here(pool) == seeded["readback"] @@ -389,15 +415,39 @@ async def test_concurrent_boots_and_migrate_serialize_on_a_published_ledger( finally: for item in pools: await item.close() - assert await _versions(pool) == ["0001", "0002", "0003", "0004", "0005"] + assert await _versions(pool) == _shipped() async with pool.connection() as connection: row = await ( await connection.execute("SELECT count(*) FROM adcp_schema_migrations") ).fetchone() - assert row[0] == 5 # each version recorded once across the four runners + assert row[0] == len(_shipped()) # each version recorded once across the four runners assert_evidence_unchanged(seeded["image"], await _image_here(pool)) +async def test_a_notifications_enabled_published_ledger_upgrades_through_every_migration( + published: tuple[str, Path], workspace: Path +) -> None: + """Status notifications add the checkpoint tables that migration 0006 re-keys.""" + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from adcp_reporting_storage.migrations import migrate + + from tests.conformance.reporting import _published_ledger_seed as seed + + _, python = published + async with isolated_reporting_pool(autocommit=True) as pool: + seeded = await _seed(python, pool, workspace, notifications=True) + assert "reporting_status_checkpoints" in seeded["image"] + store = PgReportingLedgerStore(pool=pool, notifications=True) + await seed.create_schema(pool, store, notifications=True) # a seller's boot + assert await _versions(pool) == LEDGER_VERSIONS + applied = await migrate(pool) + assert [item.version for item in applied] == _rest() + assert await _versions(pool) == _shipped() + assert await migrate(pool) == [] + assert_evidence_unchanged(seeded["image"], await _image_here(pool)) + assert await _readback_here(pool) == seeded["readback"] + + async def test_adcp_b4_fails_on_the_published_storage_and_upgrades_with_this_one( published: tuple[str, Path], b4_with_published_storage: Path, @@ -420,4 +470,4 @@ async def test_adcp_b4_fails_on_the_published_storage_and_upgrades_with_this_one assert "error" not in upgraded, upgraded["error"] assert upgraded["readback"] == seeded["readback"] assert_evidence_unchanged(seeded["image"], upgraded["image"]) - assert await _versions(pool) == ["0001", "0002", "0003"] + assert await _versions(pool) == LEDGER_VERSIONS From 49cd88b7d92fb6afee885508eac856fa48d81525 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 10:50:58 +0000 Subject: [PATCH 5/5] style(reporting-storage): wrap the create_schema migration comment Co-Authored-By: Claude Fable 5.1 --- .../src/adcp_reporting_storage/ledger/pg.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 53cc6bbe6..69f22bf8c 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -702,8 +702,9 @@ async def _create_schema_on(self, connection: Any) -> None: # and carries its version; the idempotent files below then only # re-verify what they installed. Migrations of other stores' tables # (0004 subscription state, 0005 inline storage, 0006 status checkpoint - # offerings) stay pending for python -m adcp migrate: the catalog contract of this store's older - # binaries is exactly its own reporting_* objects. + # offerings) stay pending for python -m adcp migrate: the catalog + # contract of this store's older binaries is exactly its own + # reporting_* objects. package = Path(__file__).resolve().parents[1] await apply_pending( connection,