feat(reporting-storage): make the contract-2 ledger upgrade rolling-deploy safe with expand and contract

MIGRATION_v9_to_v10.md

@@ -73,12 +73,19 @@ adcp 10.0.0b0–b3 are 1):
   such check and still install beside 0.2.0b0, which then fails every reporting
   turn, so install `adcp>=10.0.0b7`
   ([#1837](https://github.com/adcontextprotocol/adcp-client-python/issues/1837)).
-- 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. A
-  default rolling update does not: once the first new pod upgrades the ledger,
-  old pods can no longer record generations or close periods, and cannot start
-  reporting if they restart. There is no rollback short of restoring a backup.
-  See [Deploying the upgrade](docs/reporting-ledger-migration.md#deploying-the-upgrade)
-  for a Recreate or two-phase rollout.
+- An existing SDK 9 ledger is upgraded by `create_schema()` or `python -m adcp
+  migrate`, and a default rolling update works for the ledger, the
+  notification outbox and status notifications (not yet for the materializer,
+  production and status projection extensions: stop their SDK 9 workers
+  first). The upgrade first *expands*
+  the ledger: `offering_id` is added with `''` for every existing
+  account-level generation, obligation and status, the keys SDK 9 uses keep
+  their names beside the widened ones, and the applied schema versions are
+  recorded. SDK 9 pods keep starting, reading and writing through the rollout,
+  and a rollback works. Once every SDK 9 reporting process has stopped, run
+  `python -m adcp migrate --contract`: reporting subscriptions need it (before
+  it they raise `REPORTING_UPGRADE_NOT_CONTRACTED`), and after it SDK 9 can no
+  longer run against the ledger. Plain `python -m adcp migrate` never
+  contracts. Stop every SDK 9 process before `--contract`, and retry
+  `--contract` if it times out against busy new pods (it is atomic). See
+  [Deploying the upgrade](docs/reporting-ledger-migration.md#deploying-the-upgrade).

packages/adcp-reporting-storage/README.md

@@ -82,24 +82,41 @@ Install `adcp>=10.0.0b7` with this release.
 
 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.
-A default rolling update does not stop them. Old pods' reporting writes fail
-once the first new pod has upgraded the ledger, and an old pod cannot start
-reporting again. Downgrading is unsupported: restore the backup taken before
-the upgrade. See
-[Deploying the upgrade](https://github.com/adcontextprotocol/adcp-client-python/blob/next/docs/reporting-ledger-migration.md#deploying-the-upgrade)
-for a Recreate or two-phase rollout.
+existing ledger, and **a default rolling update works for the ledger, the
+notification outbox and status notifications** (the materializer, production
+and status projection extensions still need their earlier-release workers
+stopped first, below). The upgrade is an
+expand/contract pair:
+
+- **Expand** (migrations `0002` and `0006`, online-safe): `offering_id` is
+  added to `reporting_configurations`, `reporting_obligations`,
+  `reporting_consumer_statuses` and the managed-delivery tables, with `''` for
+  every existing account-level generation and `CHECK (offering_id = '')`. The
+  keys and references the earlier release uses keep their names, and the
+  widened ones are added beside them. Processes of the earlier release keep
+  starting, reading and writing account-level generations (with status
+  notifications too) while the new release rolls out, and a rollback works.
+  No evidence row is rewritten or dropped. 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).
+  Each step is transactional, serialized under the reporting schema advisory
+  lock, and a no-op when re-run.
+- **Contract** (migration `0007`, stopped-worker): run `python -m adcp migrate
+  --contract` once every earlier-release reporting process has stopped. Plain
+  `python -m adcp migrate` and `create_schema()` never contract, so the former
+  is safe as a pre-deploy job. The contracted ledger is exactly a new ledger's
+  catalog. Reporting subscriptions need it: before it, recording a
+  subscription's generation raises `ReportingUpgradeNotContractedError` (code
+  `REPORTING_UPGRADE_NOT_CONTRACTED`). After it, the earlier release cannot
+  start against the ledger, so rolling back means restoring the backup.
+
+`postgres_storage(...)` starts on a ledger whose only pending migration is the
+contract step. A new ledger is created contracted. A ledger that 0.2.0b1 or
+0.2.0b2 already widened in place counts as contracted. The materializer,
+production and status projection extensions still widen their own tables when
+the new release starts: stop their earlier-release workers first. See
+[Deploying the upgrade](https://github.com/adcontextprotocol/adcp-client-python/blob/next/docs/reporting-ledger-migration.md#deploying-the-upgrade).
 
 A ledger written by 0.1.0b5 (the `(xid, seq)` account change feed, 255-character
 row binding IDs, the orphan collector fence) upgrades the same way, with no

packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py

@@ -196,6 +196,9 @@
     from adcp_reporting_storage.migration import (
         require_owned_schema_or_empty as require_owned_schema_or_empty,
     )
+    from adcp_reporting_storage.migrations import (
+        ReportingUpgradeNotContractedError as ReportingUpgradeNotContractedError,
+    )
     from adcp_reporting_storage.outbox.pg import PgReportingOutbox as PgReportingOutbox
     from adcp_reporting_storage.outbox.status_activity_pg import (
         PgReportingActivityUnionStore as PgReportingActivityUnionStore,
@@ -270,6 +273,8 @@
     # Caller-ownership migration
     "ReportingOwnershipBackfill": ("migration", "ReportingOwnershipBackfill"),
     "ReportingOwnershipMigrationError": ("migration", "ReportingOwnershipMigrationError"),
+    # Expand/contract upgrade of an earlier release's ledger (#1856)
+    "ReportingUpgradeNotContractedError": ("migrations", "ReportingUpgradeNotContractedError"),
     "backfill_legacy_reporting": ("migration", "backfill_legacy_reporting"),
     "legacy_generation_digest": ("migration", "legacy_generation_digest"),
     "migrate_legacy_reporting": ("migration", "migrate_legacy_reporting"),

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py

@@ -161,6 +161,7 @@
     ReportingRowStorageV1,
     ReportingRowStoreError,
 )
+from adcp_reporting_storage.migrations import upgrade_not_contracted_error
 
 if TYPE_CHECKING:
     from psycopg_pool import AsyncConnectionPool
@@ -511,12 +512,15 @@ async def create_schema(self) -> None:
         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.
+        0.1.0b4 or 0.2.0b0 is expanded: ``offering_id`` is added with ``''``
+        for every existing account-level generation, beside the keys that
+        release uses, so its processes keep running through a rolling deploy.
+        The applied versions are recorded, so ``python -m adcp migrate
+        --status`` reads them. Re-running is a no-op. A boot never applies the
+        contract step (``python -m adcp migrate --contract``). The later
+        migrations (``0004`` onward: subscription state, inline storage, status
+        checkpoint offerings, the contract step) 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
@@ -1057,14 +1061,26 @@ async def _lock_account(connection: Any, account_id: str) -> None:
 
     # -- configurations ---------------------------------------------------
 
+    @staticmethod
+    async def _insert_configuration(connection: Any, query: str, params: tuple[Any, ...]) -> Any:
+        try:
+            return await connection.execute(query, params)
+        except Exception as error:
+            # An expanded ledger refuses a subscription generation until contract.
+            not_contracted = upgrade_not_contracted_error(error)
+            if not_contracted is not None:
+                raise not_contracted from error
+            raise
+
     async def put_configuration(self, configuration: ReportingConfiguration) -> None:
         reject_reserved_authoritative_party(configuration)
         key = configuration.generation_key
         payload = _configuration_payload(configuration)
         digest = _fingerprint(payload)
         async with self._connection() as connection, connection.transaction():
             await self._lock_account(connection, key.account_id)
-            inserted_cursor = await connection.execute(
+            inserted_cursor = await self._insert_configuration(
+                connection,
                 "INSERT INTO reporting_configurations"
                 " (delivery_config_id, delivery_config_version, account_id,"
                 "  report_definition_id, reporting_profile, feed_purpose, required_finality,"
@@ -3627,6 +3643,9 @@ def _translate_integrity_error(error: Exception) -> LedgerConflictError:
     Matching the index name rather than the driver's prose: the message text
     varies by Postgres version and locale, the index name does not.
     """
+    not_contracted = upgrade_not_contracted_error(error)
+    if not_contracted is not None:
+        return not_contracted
     text = str(error)
     if "reporting_revisions_one_official" in text:
         return LedgerConflictError(

packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger.sql

@@ -51,7 +51,12 @@ CREATE TABLE IF NOT EXISTS reporting_configurations (
     -- expiry instead of wedging the period forever.
     lease_worker_id         TEXT COLLATE "C",
     lease_expires_at        TIMESTAMPTZ,
-    PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version)
+    -- '' for an account-level generation; a reporting subscription keeps one
+    -- generation per resolved offering of a version. A new ledger is created
+    -- on the widened key; reporting_ledger_account_generations.sql upgrades
+    -- one created by an earlier release.
+    offering_id             TEXT COLLATE "C" NOT NULL DEFAULT '',
+    PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id)
 );
 
 CREATE INDEX IF NOT EXISTS reporting_configurations_account_idx
@@ -129,15 +134,28 @@ CREATE TABLE IF NOT EXISTS reporting_obligations (
     created_at              TIMESTAMPTZ      NOT NULL,
     -- NULL is unknown legacy currency, never an implicit USD default. The
     -- currency migration installs validation and immutability on all schemas.
-    currency                TEXT COLLATE "C"
+    currency                TEXT COLLATE "C",
+    offering_id             TEXT COLLATE "C" NOT NULL DEFAULT ''
 );
 
 -- One obligation per logical period. Without this, two workers racing a period
 -- close could commit two obligations and a seller could quietly publish twice
 -- and pick a winner.
-CREATE UNIQUE INDEX IF NOT EXISTS reporting_obligations_period_key
-    ON reporting_obligations
-       (account_id, consumer_id, delivery_config_id, delivery_config_version, period_start, period_end);
+--
+-- A new ledger keys it by offering_id too. An earlier release's ledger keeps
+-- its six-column key here (PostgreSQL resolves the columns of CREATE INDEX IF
+-- NOT EXISTS before the name): reporting_ledger_account_generations.sql
+-- upgrades it.
+DO $obligation_period_key$
+BEGIN
+    IF to_regclass(quote_ident(current_schema()) || '.reporting_obligations_period_key') IS NULL THEN
+        CREATE UNIQUE INDEX reporting_obligations_period_key
+            ON reporting_obligations
+               (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id,
+                period_start, period_end);
+    END IF;
+END
+$obligation_period_key$;
 
 CREATE INDEX IF NOT EXISTS reporting_obligations_account_idx
     ON reporting_obligations (account_id, period_end DESC);
@@ -274,6 +292,7 @@ CREATE TABLE IF NOT EXISTS reporting_consumer_statuses (
     seller_ledger_as_of     TIMESTAMPTZ,
     superseded              BOOLEAN          NOT NULL DEFAULT FALSE,
     content_sha256          TEXT COLLATE "C" NOT NULL,
+    offering_id             TEXT COLLATE "C" NOT NULL DEFAULT '',
     PRIMARY KEY (account_id, consumer_id, reporting_status_id)
 );