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-ledger-migration.md b/docs/reporting-ledger-migration.md index 776b51791..6a98f6ec3 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -281,3 +281,38 @@ 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` (`create_schema()` +applies the ledger's own `0001`–`0003`; `python -m adcp migrate` also applies +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 +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/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/packages/adcp-reporting-storage/README.md b/packages/adcp-reporting-storage/README.md index d1b2a5c6c..63d542e15 100644 --- a/packages/adcp-reporting-storage/README.md +++ b/packages/adcp-reporting-storage/README.md @@ -53,10 +53,40 @@ 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` (`create_schema()` applies +and records the ledger's own `0001`–`0003`; `python -m adcp migrate` also +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 +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..08c291ca9 --- /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 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..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 @@ -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,18 @@ 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 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 the applied + versions are recorded, so ``python -m adcp migrate --status`` reads + 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 pool, so a second process cannot observe a partially upgraded schema. @@ -666,9 +684,10 @@ 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) - for path in ( + owned = ( _DDL_PATH, _ACCOUNT_GENERATIONS_DDL_PATH, _CURRENCY_DDL_PATH, @@ -677,7 +696,21 @@ 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, 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( + 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 ab5250ca8..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 @@ -18,6 +18,12 @@ 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 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 the later ones (``0004`` onward), which only ``postgres_storage`` +and the notification stores need. """ from __future__ import annotations @@ -34,6 +40,7 @@ "Migration", "MigrationStatus", "applied_migrations", + "apply_pending", "migrate", "migration_status", "migrations", @@ -179,29 +186,64 @@ 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, *, 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) + applied = await applied_migrations(connection) + result: 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 + + +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 +253,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/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/conformance/reporting/_published_ledger_seed.py b/tests/conformance/reporting/_published_ledger_seed.py new file mode 100644 index 000000000..964d61a8b --- /dev/null +++ b/tests/conformance/reporting/_published_ledger_seed.py @@ -0,0 +1,227 @@ +"""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. 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. +""" + +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 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, notifications=notifications) + result = {"action": settings["action"]} + try: + if settings["action"] in {"seed", "upgrade"}: + await create_schema(pool, store, notifications=notifications) + 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..aab61c5ac --- /dev/null +++ b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py @@ -0,0 +1,473 @@ +"""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 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. + +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", + # Migration 0006 (status checkpoint offerings) re-keys these the same way. + "reporting_status_checkpoints", + "reporting_status_events", +} +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, *, 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, *, 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) + ) + 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}" + + +#: 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 + + statuses = await migration_status(pool) + assert [item.migration.version for item in statuses] == _shipped() + 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_reporting_storage.migrations import migrate + + 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() + + # 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) == LEDGER_VERSIONS + after = await _image_here(pool) + assert_evidence_unchanged(seeded["image"], after) + assert await _readback_here(pool) == seeded["readback"] + upgraded = await _catalog(pool) + + await store.create_schema() # re-runnable: nothing changes + assert await _image_here(pool) == after + assert await _versions(pool) == LEDGER_VERSIONS + + applied = await migrate(pool) + 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: + await PgReportingLedgerStore(pool=fresh).create_schema() + assert upgraded == await _catalog(fresh) + + +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] == _shipped() + 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) == _shipped() + async with pool.connection() as connection: + row = await ( + await connection.execute("SELECT count(*) FROM adcp_schema_migrations") + ).fetchone() + 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, + 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) == LEDGER_VERSIONS 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" 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 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())