symfonic.memory.backends.postgres_erase_ownership

postgres_erase_ownership

Indexed owner proof for explicit tenant retirement (CORE-426).

Each recursive step is one LIMIT 1 seek past a complete owner key: forty thousand generations at one owner cost one key, not forty thousand rows. An outer limit stops recursion after BUDGET+1 keys; it never sorts or hashes the history. Index probes pin custom and generic plans to those seeks. The first proof holds the lifecycle and EXCLUSIVE graph locks; the permanent job fences all later native writers. SHARE UPDATE EXCLUSIVE locks prevent concurrent DDL from removing the proof's required indexes while it runs.

This bounds logical keys/probes, not physical I/O through dead MVCC entries. The existing forget deadline is still authoritative and fails closed.

auxiliary_gate async

auxiliary_gate(conn: Any, schema: str) -> None

Refuse stores outside the separately attested bounded auxiliary subset.

Ordinary graph-only forget does not call this retirement-only gate. Empty auxiliary tables also refuse: their writers can recreate state after permanent intent because their retirement fence is not proven. The explicit schema prevents unrelated schemas/search paths from withdrawing the bound graph's retirement admission.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

async def auxiliary_gate(conn: Any, schema: str) -> None:
    """Refuse stores outside the separately attested bounded auxiliary subset.

    Ordinary graph-only forget does not call this retirement-only gate.
    Empty auxiliary tables also refuse: their writers can recreate state
    after permanent intent because their retirement fence is not proven.
    The explicit schema prevents unrelated schemas/search paths from
    withdrawing the bound graph's retirement admission.
    """
    from symfonic.memory.backends.postgres_auxiliary_schema import TABLES as supported
    from symfonic.memory.backends.postgres_journal_base_schema import TABLES as journal
    from symfonic.memory.backends.postgres_staged_auxiliary_schema import TABLES as staged

    table = await conn.fetchval(
        "SELECT c.relname FROM pg_catalog.pg_class c "
        "JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace "
        "WHERE n.nspname=$1 AND c.relname=ANY($2::text[]) ORDER BY c.relname LIMIT 1",
        schema,
        [table for table in AUXILIARY_TABLES if table not in supported
         and table not in journal and table not in staged],
    )
    if table is not None:
        raise MemoryContractError(f"{AUXILIARY_UNPROVEN}: {table}")

legacy_progress async

legacy_progress(conn: Any, table: str, tenant: str) -> bool

Exact positional root and its complete-level descendants, in two seeks.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

async def legacy_progress(conn: Any, table: str, tenant: str) -> bool:
    """Exact positional root and its complete-level descendants, in two seeks."""
    root = "tenant\x1f" + tenant
    return bool(
        await conn.fetchval(
            f'SELECT COALESCE((SELECT scope_path FROM {table} WHERE scope_path COLLATE "C">=$1 '
            'ORDER BY scope_path COLLATE "C" LIMIT 1)=$1, false)',
            root,
        )
        or await conn.fetchval(
            range_sql(table, progress=True),
            root + "\x1e",
            root + "\x1f",
        )
    )

owners_sql

owners_sql(table: str, *, own: bool = False, progress: bool = False) -> str

Exact production loose-index query; live tests EXPLAIN this SQL.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

def owners_sql(table: str, *, own: bool = False, progress: bool = False) -> str:
    """Exact production loose-index query; live tests EXPLAIN this SQL."""
    keys = (
        ("tenant_id", "scope_path")
        if own
        else (("scope_path",) if progress else ("scope_path", "tenant_id"))
    )
    ordered = ", ".join(f'{key} COLLATE "C"' for key in keys)
    previous = ", ".join(f'p.{key} COLLATE "C"' for key in keys)
    selected = ", ".join(keys)
    after = (
        f'{keys[0]} COLLATE "C">p.{keys[0]} COLLATE "C"'
        if len(keys) == 1
        else f"({ordered})>({previous})"
    )
    bounds = (
        'tenant_id COLLATE "C">=$1 AND tenant_id COLLATE "C"<=$1 AND scope_path IS NOT NULL'
        if own
        else 'scope_path COLLATE "C">=$1 AND scope_path COLLATE "C"<$2'
    )
    return (
        "WITH RECURSIVE owners AS ((SELECT " + selected + f" FROM {table} WHERE {bounds} "
        f"ORDER BY {ordered} LIMIT 1) UNION ALL SELECT n.* FROM owners p "
        f"CROSS JOIN LATERAL (SELECT {selected} FROM {table} WHERE {bounds} AND {after} "
        f"ORDER BY {ordered} LIMIT 1) n) SELECT {selected} FROM owners LIMIT {BUDGET + 1}"
    )

preflight async

preflight(conn: Any, jobs: str, scope: Any) -> None

Refuse all detectable unerasable ownership before permanent intent.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

async def preflight(conn: Any, jobs: str, scope: Any) -> None:
    """Refuse all detectable unerasable ownership before permanent intent."""
    from symfonic.memory.backends.postgres_native_subtree import _descends

    schema, tenant, prefix = jobs.rsplit(".", 1)[0], scope.tenant_id, scope.subtree_prefix
    upper = prefix[:-1] + "0"
    await auxiliary_gate(conn, schema)
    from symfonic.memory.backends import postgres_auxiliary_retirement as auxiliary

    await auxiliary.preflight(conn, schema, scope)
    from symfonic.memory.backends import postgres_journal_base_retirement as journal

    await journal.preflight(conn, schema, scope)
    if await staged_installed(conn, schema):
        from symfonic.memory.backends import postgres_staged_auxiliary_retirement as staged

        await staged.preflight(conn, schema, scope)
    await require_indexes(conn, schema)
    await vectors_gate(conn, schema, scope)
    async with index_probes(conn, ordered=True, presorted=True):
        for name in TABLES:
            table = f"{schema}.{name}"
            if await conn.fetchval(
                f"SELECT COALESCE((SELECT scope_path IS NULL FROM {table} "
                'WHERE tenant_id COLLATE "C">=$1 AND tenant_id COLLATE "C"<=$1 '
                'ORDER BY tenant_id COLLATE "C" DESC, scope_path COLLATE "C" DESC LIMIT 1), false)',
                tenant,
            ):
                raise MemoryContractError(f"native retirement residual tenant data: {name}")
            owned = await conn.fetch(owners_sql(table, own=True), tenant)
            crossed = await conn.fetch(owners_sql(table), prefix, upper)
            if len(owned) > BUDGET or len(crossed) > BUDGET:
                raise MemoryContractError("native retirement preflight ownership budget exceeded")
            if any(not _descends(row["scope_path"], scope) for row in owned) or any(
                row["tenant_id"] != tenant or not _descends(row["scope_path"], scope)
                for row in crossed
            ):
                raise MemoryContractError(f"native retirement residual tenant data: {name}")
        progress = f"{schema}.{PROGRESS}"
        rows = await conn.fetch(owners_sql(progress, progress=True), prefix, upper)
        if len(rows) > BUDGET or any(not _descends(row["scope_path"], scope) for row in rows):
            raise MemoryContractError(f"native retirement residual tenant data: {PROGRESS}")
        if await legacy_progress(conn, progress, tenant):
            raise MemoryContractError(f"native retirement residual tenant data: {PROGRESS}")

range_sql

range_sql(table: str, *, progress: bool = False) -> str

A scalar preserves ORDER BY; EXISTS may discard it and scan another index.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

def range_sql(table: str, *, progress: bool = False) -> str:
    """A scalar preserves ORDER BY; EXISTS may discard it and scan another index."""
    order = 'scope_path COLLATE "C"' + ("" if progress else ', tenant_id COLLATE "C"')
    return (
        f'SELECT (SELECT scope_path FROM {table} WHERE scope_path COLLATE "C">=$1 '
        f'AND scope_path COLLATE "C"<$2 ORDER BY {order} LIMIT 1) IS NOT NULL'
    )

require_empty async

require_empty(conn: Any, jobs: str, job_id: str) -> None

Completion/retry proof uses independent equality and range index seeks.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

async def require_empty(conn: Any, jobs: str, job_id: str) -> None:
    """Completion/retry proof uses independent equality and range index seeks."""
    tenant, prefix = await conn.fetchrow(
        f"SELECT tenant_id, prefix FROM {jobs} WHERE id=$1", job_id
    )
    schema, upper = jobs.rsplit(".", 1)[0], prefix[:-1] + "0"
    from symfonic.capabilities.memory.storage_key import StorageScopeKey
    from symfonic.memory.storage_scope import NativeStorageScope

    scope = NativeStorageScope(StorageScopeKey.parse(prefix))
    await auxiliary_gate(conn, schema)
    from symfonic.memory.backends import postgres_auxiliary_retirement as auxiliary

    await auxiliary.require_empty(conn, schema, scope)
    from symfonic.memory.backends import postgres_journal_base_retirement as journal

    await journal.require_empty(conn, schema, scope)
    if await staged_installed(conn, schema):
        from symfonic.memory.backends import postgres_staged_auxiliary_retirement as staged

        await staged.require_empty(conn, schema, scope)
    await require_indexes(conn, schema)
    await vectors_gate(conn, schema, scope, empty=True)
    async with index_probes(conn, ordered=True, presorted=True):
        for name in TABLES:
            table = f"{schema}.{name}"
            owned = await conn.fetchval(
                f"SELECT (SELECT jsonb_build_array(scope_path) FROM {table} "
                'WHERE tenant_id COLLATE "C">=$1 AND tenant_id COLLATE "C"<=$1 '
                'ORDER BY tenant_id COLLATE "C", scope_path COLLATE "C" LIMIT 1) IS NOT NULL',
                tenant,
            )
            crossed = await conn.fetchval(
                range_sql(table),
                prefix,
                upper,
            )
            if owned or crossed:
                raise MemoryContractError(f"native retirement residual tenant data: {name}")
        progress = f"{schema}.{PROGRESS}"
        if await conn.fetchval(
            range_sql(progress, progress=True),
            prefix,
            upper,
        ) or await legacy_progress(conn, progress, tenant):
            raise MemoryContractError(f"native retirement residual tenant data: {PROGRESS}")

require_indexes async

require_indexes(conn: Any, schema: str) -> None

Pin proof tables against DDL, then validate every exact index shape.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

async def require_indexes(conn: Any, schema: str) -> None:
    """Pin proof tables against DDL, then validate every exact index shape."""
    await conn.execute(
        "LOCK TABLE "
        + ", ".join(f"{schema}.{table}" for table in (*TABLES, PROGRESS))
        + " IN SHARE UPDATE EXCLUSIVE MODE"
    )
    rows = await conn.fetch(_INDEX_CHECK, schema, json.dumps(INDEXES))
    if len(rows) != len(INDEXES) or any(not row["valid"] for row in rows):
        raise MemoryContractError(UNINDEXED)

vectors_gate async

vectors_gate(conn: Any, schema: str, scope: Any = None, *, empty: bool = False) -> None

Admit only the migrated vector phase with its independent ownership proof.

Source code in src/symfonic/memory/backends/postgres_erase_ownership.py

async def vectors_gate(conn: Any, schema: str, scope: Any = None, *, empty: bool = False) -> None:
    """Admit only the migrated vector phase with its independent ownership proof."""
    if await conn.fetchval("SELECT to_regclass($1) IS NOT NULL", schema + ".memory_vectors"):
        from symfonic.memory.backends import postgres_erase_vector_proof as proof
        from symfonic.memory.backends.postgres_erase_vector_chunks import installed

        jobs = schema + ".memory_erase_jobs"
        if scope is None or not await installed(conn, jobs):
            raise MemoryContractError("native retirement cannot certify legacy memory_vectors")
        actual_schema = await conn.fetchval(
            "SELECT n.nspname FROM pg_class j JOIN pg_namespace n ON n.oid=j.relnamespace "
            "WHERE j.oid=$1::regclass", jobs,
        )
        if empty:
            await proof.require_empty(conn, actual_schema, scope)
        else:
            await proof.preflight(conn, actual_schema, scope)