From 2391f8b9b8b7b966372c9dfade526e9bb82ffce4 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 3 Nov 2026 16:26:00 +0000 Subject: [PATCH] fix(reporting-storage): check for reporting tables in the current schema only `to_regclass('name')` resolves through the whole search_path. With `search_path=tenant,public` and a ledger already in `public`, every unqualified existence check in the package could find `public`'s table while working on `tenant`: - `applied_migrations()` read `public`'s `adcp_schema_migrations`, so `python -m adcp migrate` printed "up to date" and left `tenant` unmigrated, and `postgres_storage`'s `check_schema` passed (#1838). - Migration 0002 (`reporting_ledger_account_generations.sql`) and 0006 (`reporting_status_checkpoint_offerings.sql`) took another schema's managed-delivery and status tables as present and ran their `ALTER TABLE` on the unqualified name. `tenant`'s upgrade then altered `public`'s tables, or failed with DuplicateColumn when `public` was already migrated. - The reconciliation DDL took another schema's feed as installed and skipped this schema's feed backfill. - `_period_close_generations_on` took another schema's materializer candidates as present and read them. Qualify each check with `quote_ident(current_schema())`, the pattern the rest of the package already uses, and read the recorded migrations from the current schema's table. A static test refuses any new unqualified `to_regclass('...')` in the package. Real PostgreSQL tests run `migrate` and `create_schema()` for a schema beside a migrated one on the same search_path. They check that the schema is built in place and that the other schema's catalog is unchanged. Closes #1838 Co-Authored-By: Claude Opus 5.5 --- .../src/adcp_reporting_storage/ledger/pg.py | 3 +- .../reporting_ledger_account_generations.sql | 8 +- .../reporting_ledger_reconciliation.sql | 5 +- .../reporting_status_checkpoint_offerings.sql | 6 +- .../migrations/__init__.py | 22 +++- .../test_reporting_storage_migrations_pg.py | 112 ++++++++++++++++++ 6 files changed, 143 insertions(+), 13 deletions(-) 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 69f22bf8c..faf575296 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 @@ -3301,7 +3301,8 @@ async def _period_close_generations_on( # every catalog object on the lease path. Call only after the account lock. present = await ( await connection.execute( - "SELECT to_regclass('reporting_materializer_candidates') IS NOT NULL" + "SELECT to_regclass(quote_ident(current_schema())" + " || '.reporting_materializer_candidates') IS NOT NULL" ) ).fetchone() if not present[0]: 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 250b1343c..69e23bff5 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 @@ -95,7 +95,7 @@ BEGIN -- (reconciliation records, materializer discovery and status checkpoints). -- Production generations serve account-level generations only; a -- subscription generation is refused by their CHECK. - IF to_regclass('reporting_reconciliation_records') IS NOT NULL THEN + IF to_regclass(quote_ident(current_schema()) || '.reporting_reconciliation_records') IS NOT NULL THEN ALTER TABLE reporting_reconciliation_records ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', ADD CONSTRAINT reporting_reconciliation_records_generation_fkey @@ -104,7 +104,7 @@ BEGIN REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id); END IF; - IF to_regclass('reporting_materializer_discovery') IS NOT NULL THEN + IF to_regclass(quote_ident(current_schema()) || '.reporting_materializer_discovery') IS NOT NULL THEN ALTER TABLE reporting_materializer_discovery ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', ADD CONSTRAINT reporting_materializer_discovery_generation_fkey @@ -113,7 +113,7 @@ BEGIN REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id); END IF; - IF to_regclass('reporting_status_scope_checkpoints') IS NOT NULL THEN + IF to_regclass(quote_ident(current_schema()) || '.reporting_status_scope_checkpoints') IS NOT NULL THEN ALTER TABLE reporting_status_scope_checkpoints ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '', ADD CONSTRAINT reporting_status_scope_checkpoints_generation_fkey @@ -122,7 +122,7 @@ BEGIN REFERENCES reporting_configurations(account_id, consumer_id, delivery_config_id, delivery_config_version, offering_id); END IF; - IF to_regclass('reporting_production_generations') IS NOT NULL THEN + IF to_regclass(quote_ident(current_schema()) || '.reporting_production_generations') IS NOT NULL THEN ALTER TABLE reporting_production_generations ADD COLUMN offering_id TEXT COLLATE "C" NOT NULL DEFAULT '' CONSTRAINT reporting_production_generations_account_generation diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql index 440975c5d..3641a36a7 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql @@ -829,7 +829,10 @@ BEGIN -- The optional feed has a dense head for each authenticated principal. It -- never allocates Core ledger sequences or widens Core's record vocabulary. - feed_installed := to_regclass('reporting_reconciliation_changes') IS NOT NULL; + -- Look in this schema only, so another schema's feed on the search_path + -- cannot skip this one's backfill. + feed_installed := to_regclass( + quote_ident(current_schema()) || '.reporting_reconciliation_changes') IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS reporting_reconciliation_feed_identity ON reporting_reconciliation_records(account_id, consumer_id, namespace, record_id, record_kind, change_id, content_sha256); diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql index 3bad3c22a..33e82d1f5 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_status_checkpoint_offerings.sql @@ -32,7 +32,7 @@ DECLARE events TEXT; BEGIN PERFORM pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema())); - IF to_regclass('reporting_status_scope_checkpoints') IS NULL THEN + IF to_regclass(quote_ident(current_schema()) || '.reporting_status_scope_checkpoints') IS NULL THEN RETURN; END IF; SELECT ARRAY( @@ -224,14 +224,14 @@ BEGIN RETURN NEW; END $event_guard$; - IF to_regclass('reporting_status_notification_events') IS NOT NULL AND NOT EXISTS ( + IF to_regclass(quote_ident(current_schema()) || '.reporting_status_notification_events') IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM pg_trigger WHERE tgrelid='reporting_status_notification_events'::regclass AND tgname='reporting_status_event_guard' AND NOT tgisinternal) THEN CREATE TRIGGER reporting_status_event_guard BEFORE INSERT ON reporting_status_notification_events FOR EACH ROW EXECUTE FUNCTION reporting_status_event_guard(); END IF; - IF to_regclass('reporting_projection_accounts') IS NULL THEN + IF to_regclass(quote_ident(current_schema()) || '.reporting_projection_accounts') IS NULL THEN RETURN; END IF; -- The projection's status events: the same checkpoint check, after 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 855b4e96a..013239c87 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 @@ -145,12 +145,26 @@ def migrations() -> tuple[Migration, ...]: async def applied_migrations(connection: Any) -> dict[str, datetime]: - """``version`` -> ``applied_at`` for every recorded migration (empty when none).""" - cursor = await connection.execute("SELECT to_regclass('adcp_schema_migrations') IS NOT NULL") + """``version`` -> ``applied_at`` for every recorded migration (empty when none). + + Reads the ``current_schema()`` table only: an unqualified name resolves + through the whole ``search_path``, so another schema's record would + otherwise report this one as migrated. + """ + from psycopg import sql + + cursor = await connection.execute( + "SELECT current_schema()," + " to_regclass(quote_ident(current_schema()) || '.adcp_schema_migrations') IS NOT NULL" + ) row = await cursor.fetchone() - if not row or not row[0]: + if not row or not row[1]: return {} - cursor = await connection.execute("SELECT version, applied_at FROM adcp_schema_migrations") + cursor = await connection.execute( + sql.SQL("SELECT version, applied_at FROM {}").format( + sql.Identifier(row[0], "adcp_schema_migrations") + ) + ) return {str(version): applied_at for version, applied_at in await cursor.fetchall()} diff --git a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py index 199d6016a..4a2d26648 100644 --- a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py @@ -2,6 +2,8 @@ from __future__ import annotations +from contextlib import asynccontextmanager + import pytest from adcp.reporting._errors import ReliableReportingConfigurationError @@ -146,3 +148,113 @@ async def test_a_managed_service_runs_on_postgres_storage() -> None: assert await engine.list_subscription_principals() == [PrincipalRef(None, AGENCY)] finally: await service.close() + + +def test_no_storage_existence_check_resolves_through_the_search_path() -> None: + """#1838: ``to_regclass('name')`` finds a table in any schema on the path. + + An existence check that does would read, skip or alter another schema's + tables; qualify it with ``quote_ident(current_schema())``. + """ + import re + from pathlib import Path + + import adcp_reporting_storage + + root = Path(adcp_reporting_storage.__file__).parent + unqualified = re.compile(r"to_regclass\(\s*'") + found = [ + f"{path.relative_to(root)}:{number}" + for path in sorted(root.rglob("*")) + if path.suffix in {".py", ".sql"} + for number, line in enumerate(path.read_text().splitlines(), 1) + if unqualified.search(line) + ] + assert found == [] + + +async def _schema(pool) -> str: + async with pool.connection() as connection: + return (await (await connection.execute("SELECT current_schema()")).fetchone())[0] + + +async def _catalog(pool, schema: str) -> list[tuple]: + """Columns, constraints and recorded migrations of ``schema``.""" + async with pool.connection() as connection: + columns = await ( + await connection.execute( + "SELECT table_name, column_name, data_type FROM information_schema.columns" + " WHERE table_schema = %s ORDER BY 1, 2", + (schema,), + ) + ).fetchall() + constraints = await ( + await connection.execute( + "SELECT c.conrelid::regclass::text, c.conname, pg_get_constraintdef(c.oid)" + " FROM pg_constraint c JOIN pg_namespace n ON n.oid = c.connamespace" + " WHERE n.nspname = %s ORDER BY 1, 2", + (schema,), + ) + ).fetchall() + recorded = await ( + await connection.execute("SELECT version FROM adcp_schema_migrations ORDER BY 1") + ).fetchall() + return [*columns, *constraints, *recorded] + + +@asynccontextmanager +async def _search_path(*schemas: str): + import os + + from psycopg_pool import AsyncConnectionPool + + async with AsyncConnectionPool( + os.environ["ADCP_PG_TEST_URL"], + kwargs={"options": f"-csearch_path={','.join(schemas)}", "autocommit": True}, + min_size=1, + max_size=2, + open=False, + ) as pool: + await pool.wait(timeout=10) + yield pool + + +@pytest.mark.asyncio +async def test_migrate_upgrades_its_own_schema_beside_a_migrated_one() -> None: + """#1838: a migrated schema later on the search_path is not this schema's record.""" + from adcp_reporting_storage.migrations import migrate, migrations, pending_migrations + + every = [item.version for item in migrations()] + async with ( + isolated_reporting_pool(autocommit=True) as other, + isolated_reporting_pool(autocommit=True) as own, + ): + await migrate(other) + other_schema, own_schema = await _schema(other), await _schema(own) + before = await _catalog(other, other_schema) + async with _search_path(own_schema, other_schema) as shared: + assert [item.version for item in await pending_migrations(shared)] == every + assert [item.version for item in await migrate(shared)] == every + assert await pending_migrations(shared) == [] + assert await pending_migrations(own) == [] + assert await _catalog(own, own_schema) == before # the same schema, built in place + assert await _catalog(other, other_schema) == before + + +@pytest.mark.asyncio +async def test_create_schema_builds_its_own_schema_beside_a_migrated_one() -> None: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from adcp_reporting_storage.migrations import migrate, pending_migrations + + async with ( + isolated_reporting_pool(autocommit=True) as other, + isolated_reporting_pool(autocommit=True) as own, + ): + await migrate(other) + other_schema, own_schema = await _schema(other), await _schema(own) + before = await _catalog(other, other_schema) + async with _search_path(own_schema, other_schema) as shared: + await PgReportingLedgerStore(pool=shared).create_schema() + # create_schema() records its own 0001-0003; the rest stay for ``migrate``. + assert [item.version for item in await pending_migrations(own)] == ["0004", "0005", "0006"] + assert await _catalog(other, other_schema) == before