bokelley · Nov 3, 2026
@@ -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]:

@@ -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

@@ -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);

@@ -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

@@ -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()}
 
 

@@ -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