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 0eef67eb4c..59d4a48f9f 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py @@ -168,6 +168,9 @@ from adcp_reporting_storage.migration import ( ReportingOwnershipMigrationError as ReportingOwnershipMigrationError, ) + from adcp_reporting_storage.migration import ( + ReportingSchemaAheadError as ReportingSchemaAheadError, + ) from adcp_reporting_storage.migration import ( backfill_legacy_reporting as backfill_legacy_reporting, ) @@ -252,6 +255,7 @@ # Caller-ownership migration "ReportingOwnershipBackfill": ("migration", "ReportingOwnershipBackfill"), "ReportingOwnershipMigrationError": ("migration", "ReportingOwnershipMigrationError"), + "ReportingSchemaAheadError": ("migration", "ReportingSchemaAheadError"), "backfill_legacy_reporting": ("migration", "backfill_legacy_reporting"), "legacy_generation_digest": ("migration", "legacy_generation_digest"), "migrate_legacy_reporting": ("migration", "migrate_legacy_reporting"), 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 c84109c437..f22c48ed0e 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 @@ -159,6 +159,7 @@ ReportingRowStorageV1, ReportingRowStoreError, ) +from adcp_reporting_storage.migration import schema_ahead_error if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool @@ -1028,40 +1029,46 @@ async def put_configuration(self, configuration: ReportingConfiguration) -> None digest = _fingerprint(payload) async with self._connection() as connection, connection.transaction(): await self._lock_account(connection, key.account_id) - inserted_cursor = await connection.execute( - "INSERT INTO reporting_configurations" - " (delivery_config_id, delivery_config_version, account_id," - " report_definition_id, reporting_profile, feed_purpose, required_finality," - " account_timezone, schedule, media_buy_ids, activated_at, deactivated_at," - " automated_recovery_seconds, status_retention_days, definition," - " authoritative_party, content_sha256, consumer_id, quarantined)" - " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s, %s, %s," - " %s::jsonb, %s, %s, %s, %s)" - " ON CONFLICT (account_id, consumer_id, delivery_config_id, delive" - "ry_config_version) DO NOTHING" - " RETURNING delivery_config_id", - ( - key.delivery_config_id, - key.delivery_config_version, - key.account_id, - configuration.report_definition_id, - configuration.reporting_profile, - configuration.feed_purpose, - configuration.required_finality, - configuration.account_timezone, - _json(payload["schedule"]), - _json(sorted(configuration.media_buy_ids)), - configuration.activated_at, - configuration.deactivated_at, - configuration.automated_recovery_window.total_seconds(), - configuration.status_retention_days, - _json(payload["definition"]) if payload["definition"] else None, - configuration.authoritative_party, - digest, - configuration.consumer_id, - configuration.quarantined, - ), - ) + try: + inserted_cursor = await connection.execute( + "INSERT INTO reporting_configurations" + " (delivery_config_id, delivery_config_version, account_id," + " report_definition_id, reporting_profile, feed_purpose, required_finality," + " account_timezone, schedule, media_buy_ids, activated_at, deactivated_at," + " automated_recovery_seconds, status_retention_days, definition," + " authoritative_party, content_sha256, consumer_id, quarantined)" + " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s, %s, %s," + " %s::jsonb, %s, %s, %s, %s)" + " ON CONFLICT (account_id, consumer_id, delivery_config_id, delive" + "ry_config_version) DO NOTHING" + " RETURNING delivery_config_id", + ( + key.delivery_config_id, + key.delivery_config_version, + key.account_id, + configuration.report_definition_id, + configuration.reporting_profile, + configuration.feed_purpose, + configuration.required_finality, + configuration.account_timezone, + _json(payload["schedule"]), + _json(sorted(configuration.media_buy_ids)), + configuration.activated_at, + configuration.deactivated_at, + configuration.automated_recovery_window.total_seconds(), + configuration.status_retention_days, + _json(payload["definition"]) if payload["definition"] else None, + configuration.authoritative_party, + digest, + configuration.consumer_id, + configuration.quarantined, + ), + ) + except Exception as error: + ahead = schema_ahead_error(error) + if ahead is not None: + raise ahead from error + raise inserted = await inserted_cursor.fetchone() # Check *after* the insert. ON CONFLICT waits for a concurrent # winner; a fresh READ COMMITTED statement sees its retained @@ -3493,6 +3500,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. """ + ahead = schema_ahead_error(error) + if ahead is not None: + return LedgerConflictError("REPORTING_SCHEMA_AHEAD", str(ahead)) text = str(error) if "reporting_revisions_one_official" in text: return LedgerConflictError( diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql index 764c75163d..18296264a5 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_account_generations.sql @@ -29,6 +29,13 @@ BEGIN ] THEN RETURN; END IF; + -- adcp-reporting-storage 0.2+ widens the key with offering_id (storage + -- contract 2). This release cannot run on it, and the caller-ownership + -- remedy below would be wrong advice. + IF 'offering_id' = ANY(primary_key_columns) THEN + RAISE EXCEPTION 'adcp-reporting-storage 0.2 or later has upgraded this reporting ledger to storage contract 2; this release implements storage contract 1' + USING HINT = 'Run adcp>=10.0.0b7 with adcp-reporting-storage>=0.2.0b1 on every reporting process. Downgrading is unsupported: restore the backup taken before the upgrade. Do not run migrate_legacy_reporting.'; + END IF; RAISE EXCEPTION 'Reporting caller ownership requires a stopped-worker maintenance migration' USING HINT = 'Use migrate_legacy_reporting; retain legacy state in quarantine and backfill authoritative ownership.'; 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 f9753f3e63..3c7d0129f3 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migration.py @@ -26,6 +26,7 @@ __all__ = [ "ReportingOwnershipBackfill", "ReportingOwnershipMigrationError", + "ReportingSchemaAheadError", "backfill_legacy_reporting", "legacy_generation_digest", "migrate_legacy_reporting", @@ -43,6 +44,41 @@ class ReportingOwnershipMigrationError(ReportingOperatorActionError): """ +_CONTRACT_2_REMEDY = ( + "Run adcp>=10.0.0b7 with adcp-reporting-storage>=0.2.0b1 on every reporting process." + " Downgrading is unsupported: to go back, restore the backup taken before the upgrade." + " This is not a caller-ownership migration; do not run migrate_legacy_reporting." +) +# PostgreSQL: no unique index or constraint matches an ON CONFLICT target. +_NO_MATCHING_CONFLICT_KEY = "42P10" + + +class ReportingSchemaAheadError(ReportingOperatorActionError): + """adcp-reporting-storage 0.2 or later upgraded this ledger (storage contract 2). + + This release implements storage contract 1: its generation, obligation + and status keys name the columns before ``offering_id`` was added, so it + can neither start on the upgraded ledger nor write to it. Service startup + re-raises it unwrapped, so the operator sees the remedy. + """ + + +def schema_ahead_error(error: BaseException) -> ReportingSchemaAheadError | None: + """The contract-2 refusal for a write whose conflict key the upgrade widened. + + A running process of this release meets an upgraded ledger as PostgreSQL + 42P10 on its ``ON CONFLICT`` writes, since the unique keys they name now + include ``offering_id``. ``None`` for any other error. + """ + if getattr(error, "sqlstate", None) != _NO_MATCHING_CONFLICT_KEY: + return None + return ReportingSchemaAheadError( + "this reporting write names a key that a newer adcp-reporting-storage (storage contract" + " 2, with offering_id) has widened, so the ledger was upgraded while this process ran. " + + _CONTRACT_2_REMEDY + ) + + @dataclass(frozen=True) class ReportingOwnershipBackfill: """Authoritative mapping with an explicit retained-only recovery decision. @@ -83,9 +119,17 @@ async def require_owned_schema_or_empty(connection: Any) -> None: "SELECT EXISTS (SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace" " 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')" + " 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')" ) ).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" + " storage contract 1 and cannot run against it. " + _CONTRACT_2_REMEDY + ) if row[0] and not row[1]: raise ReportingOwnershipMigrationError( "If all legacy tables are empty, call replace_empty_legacy_reporting. " diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py index 63f4e91f31..4b1f36a8a2 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py @@ -49,6 +49,7 @@ persist_replay_lifecycles_on, settle_snapshot_on, ) +from adcp_reporting_storage.migration import schema_ahead_error from adcp_reporting_storage.outbox.pg import PgReportingOutbox, database_now _KEY = ( @@ -288,13 +289,20 @@ async def _lock_scopes_on(self, connection: Any, snapshot: ReportingStatusSnapsh # Insert placeholders before allocating random IDs. New and existing # typed scopes have exactly the same row-lock/notification boundary. for scope in projection_scopes(snapshot): - await connection.execute( - f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, scope, fingerprint," # nosec B608 - " snapshot, source_sequence, baseline, publishable, selector_writer_floor)" - ' VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,\'{"health":"waiting"}\',0,FALSE,FALSE,2)' - f" ON CONFLICT ({_KEY}) DO NOTHING", # nosec B608 - (*scope.checkpoint_key, json.dumps(asdict(scope)), "0" * 64), - ) + try: + await connection.execute( + f"INSERT INTO reporting_status_scope_checkpoints ({_KEY}, scope, fingerprint," # nosec B608 + " snapshot, source_sequence, baseline, publishable, selector_writer_floor)" + " VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s," + ' \'{"health":"waiting"}\',0,FALSE,FALSE,2)' + f" ON CONFLICT ({_KEY}) DO NOTHING", # nosec B608 + (*scope.checkpoint_key, json.dumps(asdict(scope)), "0" * 64), + ) + except Exception as error: + ahead = schema_ahead_error(error) + if ahead is not None: + raise ahead from error + raise await ( await connection.execute( "SELECT 1 FROM reporting_status_scope_checkpoints WHERE account_id=%s" diff --git a/tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py b/tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py new file mode 100644 index 0000000000..fb69f2e2e2 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_storage_contract2_refusal_pg.py @@ -0,0 +1,130 @@ +"""A ledger upgraded to storage contract 2 is refused clearly by this release (#1831). + +adcp-reporting-storage 0.2.0b1+ upgrades a 0.1.x ledger in place: migration +0002 adds ``offering_id`` and widens the configuration key and the obligation +period key with it. If a process of this release keeps running during a +rolling deploy, or restarts afterwards, it must name the actual problem and +remedy rather than fail with a raw driver error or point at the unrelated +caller-ownership migration. + +``_widen`` applies the same key changes as 0002 on ``next`` +(``reporting_ledger_account_generations.sql``). +""" + +from __future__ import annotations + +import pytest + +from adcp.reporting.ledger import LedgerConflictError + +from ._generation_support import configuration, isolated_reporting_pool, obligation_for + +_WIDEN = """ +ALTER TABLE reporting_configurations ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''; +ALTER TABLE reporting_configurations DROP CONSTRAINT reporting_configurations_pkey CASCADE; +ALTER TABLE reporting_configurations ADD CONSTRAINT reporting_configurations_pkey + PRIMARY KEY (account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id); +ALTER TABLE reporting_obligations ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''; +DROP INDEX reporting_obligations_period_key; +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); +ALTER TABLE reporting_consumer_statuses ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT ''; +""" + + +async def _upgraded_store(pool): + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + store = PgReportingLedgerStore(pool=pool) + await store.create_schema() + await store.put_configuration(configuration("acct_a")) + async with pool.connection() as connection: + await connection.execute(_WIDEN) + return store + + +def _assert_remedy(message: str) -> None: + assert "storage contract 2" in message + assert "adcp>=10.0.0b7 with adcp-reporting-storage>=0.2.0b1" in message + assert "restore the backup" in message + assert "do not run migrate_legacy_reporting" in message + + +def test_only_a_missing_conflict_key_is_a_contract_2_refusal() -> None: + from adcp_reporting_storage.migration import schema_ahead_error + + class _Error(Exception): + def __init__(self, sqlstate: str) -> None: + super().__init__(sqlstate) + self.sqlstate = sqlstate + + assert schema_ahead_error(_Error("42P10")) is not None + assert schema_ahead_error(_Error("23505")) is None # unique violation + assert schema_ahead_error(ValueError("no sqlstate")) is None + + +@pytest.mark.asyncio +async def test_create_schema_refuses_an_upgraded_ledger_and_changes_nothing() -> None: + from adcp_reporting_storage import ReportingSchemaAheadError + + async with isolated_reporting_pool(autocommit=True) as pool: + store = await _upgraded_store(pool) + async with pool.connection() as connection: + before = await ( + await connection.execute( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint" + " WHERE conname = 'reporting_configurations_pkey'" + " AND connamespace = current_schema()::regnamespace" + ) + ).fetchall() + with pytest.raises(ReportingSchemaAheadError) as refused: + await store.create_schema() + _assert_remedy(str(refused.value)) + async with pool.connection() as connection: + after = await ( + await connection.execute( + "SELECT pg_get_constraintdef(oid) FROM pg_constraint" + " WHERE conname = 'reporting_configurations_pkey'" + " AND connamespace = current_schema()::regnamespace" + ) + ).fetchall() + assert after == before and "offering_id" in before[0][0] + + +@pytest.mark.asyncio +async def test_the_ddl_alone_names_contract_2_not_the_ownership_migration() -> None: + """For operators who apply the packaged SQL themselves.""" + from importlib.resources import files + + ddl = ( + files("adcp_reporting_storage.ledger") + .joinpath("reporting_ledger_account_generations.sql") + .read_text() + ) + async with isolated_reporting_pool(autocommit=True) as pool: # skips without psycopg + from psycopg import errors + + await _upgraded_store(pool) + async with pool.connection() as connection: + with pytest.raises(errors.RaiseException) as raised: + await connection.execute(ddl) + diagnostic = raised.value.diag + assert "storage contract 2" in (diagnostic.message_primary or "") + assert "Do not run migrate_legacy_reporting" in (diagnostic.message_hint or "") + + +@pytest.mark.asyncio +async def test_a_running_process_names_the_upgrade_on_every_widened_write() -> None: + from adcp_reporting_storage import ReportingSchemaAheadError + + async with isolated_reporting_pool(autocommit=True) as pool: + store = await _upgraded_store(pool) + with pytest.raises(ReportingSchemaAheadError) as refused: + await store.put_configuration(configuration("acct_b")) + _assert_remedy(str(refused.value)) + with pytest.raises(LedgerConflictError) as conflict: + await store.commit_obligation(obligation_for(configuration("acct_a"))) + assert conflict.value.code == "REPORTING_SCHEMA_AHEAD" + _assert_remedy(str(conflict.value)) diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index cc4edb3fe2..48f3b62c85 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -8225,6 +8225,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", + "ReportingSchemaAheadError": "adcp_reporting_storage.migration", "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",