From 198470b45a2a6358f2c5cd1fd412c63c0ecca412 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 8 Nov 2026 03:29:55 +0000 Subject: [PATCH] fix(reporting-storage): let 0.1.x run on an expanded ledger until it is contracted #1857 refused any ledger whose reporting_configurations has offering_id. The rolling-safe upgrade on next (#1856) adds that column in an expand step but keeps the four-column generation key that 0.1.x writes, until the operator contracts. So 0.1.0b5 would refuse to restart mid-rollout, though it otherwise runs on the expand state unchanged. Refuse only when the configuration primary key includes offering_id, i.e. once the ledger is contracted. That is the same test the packaged SQL's key check makes. Refs #1856 Co-Authored-By: Claude Opus 5.5 --- .../src/adcp_reporting_storage/migration.py | 11 ++++-- ..._reporting_storage_contract2_refusal_pg.py | 37 +++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py index 3c7d0129f..5f9704714 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py +++ b/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]: diff --git a/tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py b/tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py index fb69f2e2e..53047c13c 100644 --- a/tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py +++ b/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"