symfonic.memory.backends.postgres_edge_index_preflight

postgres_edge_index_preflight

Read-only catalog and data preflight for the scoped edge-index migration.

This is evidence for an operator, not a migration or a write-barrier probe. Fleet writer state cannot be inferred from one PostgreSQL connection.

DirectEdgeIndexNotReady

Bases: RuntimeError

The configured edge index cannot protect a direct mutation.

EdgeIndexPreflight dataclass

EdgeIndexPreflight(schema: str, indexes: tuple[EdgeIndexMetadata, ...], old_index_present: bool, new_index_present: bool, null_scope_rows: int, duplicate_effective_owner_groups: int, sibling_tuple_groups: int, fleet_writer_state: str = 'UNKNOWN')

migration_authorized property

migration_authorized: bool

A database snapshot cannot establish the fleet write barrier.

EdgeIndexPreflightError

Bases: RuntimeError

The database cannot be inspected completely and safely.

LegacyEdgeBootstrapRefused

Bases: RuntimeError

A default host cannot restore the tenant-wide index after cutover.

inspect_edge_indexes async

inspect_edge_indexes(conn: Any, *, schema: str) -> EdgeIndexPreflight

Inspect one schema under a consistent read-only transaction.

The caller supplies an asyncpg-style connection. No row identifiers or scope values leave the database; only counts and index metadata do.

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

async def inspect_edge_indexes(conn: Any, *, schema: str) -> EdgeIndexPreflight:
    """Inspect one schema under a consistent read-only transaction.

    The caller supplies an asyncpg-style connection. No row identifiers or
    scope values leave the database; only counts and index metadata do.
    """
    if not schema or "\x00" in schema:
        raise EdgeIndexPreflightError("An explicit valid schema is required")
    try:
        async with conn.transaction(isolation="repeatable_read", readonly=True):
            table = await conn.fetchrow(_TABLE_SQL, schema)
            if table is None or table["relkind"] != "r":
                raise EdgeIndexPreflightError("memory_edges is absent or unsupported")
            if not table["can_select"] or table["relrowsecurity"] or table["relforcerowsecurity"]:
                raise EdgeIndexPreflightError("Complete edge-row visibility is not established")
            oid = table["oid"]
            column_rows = await conn.fetch(_COLUMNS_SQL, oid)
            columns = {row["attname"] for row in column_rows}
            if not columns >= _REQUIRED_COLUMNS:
                raise EdgeIndexPreflightError("memory_edges has an unknown schema")
            if any(
                row["collation_deterministic"] is not True
                for row in column_rows
                if row["attname"] in _REQUIRED_COLUMNS
            ):
                raise EdgeIndexPreflightError("Unsafe edge identity equality")
            index_rows = await conn.fetch(_INDEXES_SQL, oid)
            if any(
                row[key] is None
                for row in index_rows
                for key in (
                    "name", "definition", "unique", "valid", "ready",
                    "unsafe_identity_equality",
                )
            ):
                raise EdgeIndexPreflightError("Index metadata is incomplete")
            if any(
                row["name"] in (OLD_INDEX, NEW_INDEX) and row["unsafe_identity_equality"]
                for row in index_rows
            ):
                raise EdgeIndexPreflightError("Unsafe edge identity equality")
            indexes = tuple(
                EdgeIndexMetadata(
                    name=row["name"],
                    definition=row["definition"],
                    unique=row["unique"],
                    valid=row["valid"],
                    ready=row["ready"],
                )
                for row in index_rows
            )
            counts = await conn.fetchrow(_counts_sql(schema))
            if counts is None or any(
                counts[key] is None or counts[key] < 0
                for key in (
                    "null_scope_rows",
                    "duplicate_effective_owner_groups",
                    "sibling_tuple_groups",
                )
            ):
                raise EdgeIndexPreflightError("Edge counts are unavailable")
            return EdgeIndexPreflight(
                schema=schema,
                indexes=indexes,
                old_index_present=any(index.name == OLD_INDEX for index in indexes),
                new_index_present=any(index.name == NEW_INDEX for index in indexes),
                null_scope_rows=counts["null_scope_rows"],
                duplicate_effective_owner_groups=counts["duplicate_effective_owner_groups"],
                sibling_tuple_groups=counts["sibling_tuple_groups"],
            )
    except EdgeIndexPreflightError:
        raise
    except Exception:
        # Driver errors can include SQL values. Never put them in operator output.
        raise EdgeIndexPreflightError("Edge-index inspection failed") from None

require_direct_edge_index async

require_direct_edge_index(conn: Any, mode: str) -> None

Check the writer's resolved table on its acquired connection.

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

async def require_direct_edge_index(conn: Any, mode: str) -> None:
    """Check the writer's resolved table on its acquired connection."""
    if mode == "scoped":
        from symfonic.memory.backends.postgres_edge_scoped_writer import (
            require_scoped_edge_index,
        )
        await require_scoped_edge_index(conn)
        return
    if mode != "tenant_guarded":
        raise DirectEdgeIndexNotReady("unknown edge index mode")
    try:
        indexes = await conn.fetchrow(_BOOTSTRAP_SQL, OLD_INDEX, NEW_INDEX)
        if indexes is None or indexes["old_usable"] is not True:
            raise DirectEdgeIndexNotReady("legacy edge index is not ready")
    except DirectEdgeIndexNotReady:
        raise
    except Exception:
        raise DirectEdgeIndexNotReady("legacy edge index verification failed") from None

require_legacy_bootstrap_safe async

require_legacy_bootstrap_safe(conn: Any) -> None

Refuse default index DDL once the scoped index has replaced it.

The resolved table is the same unqualified table used by ensure_schema. A fresh database has neither index and keeps legacy bootstrap behavior.

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

async def require_legacy_bootstrap_safe(conn: Any) -> None:
    """Refuse default index DDL once the scoped index has replaced it.

    The resolved table is the same unqualified table used by ensure_schema.
    A fresh database has neither index and keeps legacy bootstrap behavior.
    """
    try:
        indexes = await conn.fetchrow(_BOOTSTRAP_SQL, OLD_INDEX, NEW_INDEX)
        if indexes is None or any(
            indexes[key] is not True and indexes[key] is not False
            for key in ("old_present", "new_present", "old_usable")
        ):
            raise LegacyEdgeBootstrapRefused("edge bootstrap catalog is unavailable")
        if indexes["new_present"] and not indexes["old_usable"]:
            raise LegacyEdgeBootstrapRefused(
                "scoped edge index is active; default edge bootstrap is refused"
            )
    except LegacyEdgeBootstrapRefused:
        raise
    except Exception:
        raise LegacyEdgeBootstrapRefused("edge bootstrap catalog check failed") from None