fix(reporting-storage): let 0.1.x run on an expanded ledger until it is contracted

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

@@ -120,14 +120,19 @@ async def require_owned_schema_or_empty(connection: Any) -> None:
             " WHERE n.nspname=current_schema() AND c.relname='reporting_configurations'),"
             " EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=current_schema()"
             " AND table_name='reporting_configurations' AND column_name='consumer_id'),"
-            " EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=current_schema()"
-            " AND table_name='reporting_configurations' AND column_name='offering_id')"
+            # The generation key, not the column: a ledger a newer release has expanded but
+            # not yet contracted carries offering_id beside the four-column key this release
+            # writes, and this release runs on it unchanged during a rolling upgrade (#1856).
+            " EXISTS (SELECT 1 FROM pg_constraint pk JOIN pg_attribute a"
+            " ON a.attrelid=pk.conrelid AND a.attnum=ANY(pk.conkey)"
+            " WHERE pk.conrelid=to_regclass(quote_ident(current_schema())"
+            " || '.reporting_configurations') AND pk.contype='p' AND a.attname='offering_id')"
         )
     ).fetchone()
     if row[2]:
         raise ReportingSchemaAheadError(
             "adcp-reporting-storage 0.2 or later has upgraded this reporting ledger to storage"
-            " contract 2 (reporting_configurations has offering_id). This release implements"
+            " contract 2 (its generation key includes offering_id). This release implements"
             " storage contract 1 and cannot run against it. " + _CONTRACT_2_REMEDY
         )
     if row[0] and not row[1]:

tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py

@@ -33,6 +33,26 @@
 ALTER TABLE reporting_consumer_statuses ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '';
 """
 
+# The ledger-core part of the expand state 0002 leaves on ``next`` (#1856): offering_id beside
+# the four-column keys this release writes, until the operator contracts.
+_EXPAND = """
+ALTER TABLE reporting_configurations
+    ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
+    ADD CONSTRAINT reporting_configurations_generation_key
+        UNIQUE (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id),
+    ADD CONSTRAINT reporting_configurations_uncontracted CHECK (offering_id = '');
+ALTER TABLE reporting_obligations
+    ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
+    ADD CONSTRAINT reporting_obligations_uncontracted CHECK (offering_id = '');
+CREATE UNIQUE INDEX reporting_obligations_offering_period_key
+    ON reporting_obligations
+       (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id,
+        period_start, period_end);
+ALTER TABLE reporting_consumer_statuses
+    ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '',
+    ADD CONSTRAINT reporting_consumer_statuses_uncontracted CHECK (offering_id = '');
+"""
+
 
 async def _upgraded_store(pool):
     from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore
@@ -128,3 +148,20 @@ async def test_a_running_process_names_the_upgrade_on_every_widened_write() -> N
             await store.commit_obligation(obligation_for(configuration("acct_a")))
         assert conflict.value.code == "REPORTING_SCHEMA_AHEAD"
         _assert_remedy(str(conflict.value))
+
+
+@pytest.mark.asyncio
+async def test_an_expanded_ledger_that_is_not_yet_contracted_still_runs_this_release() -> None:
+    """During a rolling upgrade (#1856) this release keeps booting and writing."""
+    from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore
+
+    async with isolated_reporting_pool(autocommit=True) as pool:
+        store = PgReportingLedgerStore(pool=pool)
+        await store.create_schema()
+        await store.put_configuration(configuration("acct_a"))
+        async with pool.connection() as connection:
+            await connection.execute(_EXPAND)
+        await store.create_schema()  # a restart
+        await store.put_configuration(configuration("acct_b"))
+        committed = await store.commit_obligation(obligation_for(configuration("acct_a")))
+        assert committed.account_id == "acct_a"