Skip to content

symfonic.memory.backends.postgres_graph_bootstrap

postgres_graph_bootstrap

Owner bootstrap and old-login fencing for the graph runtime (CORE-127).

Split from postgres_graph_runtime, which still re-exports both helpers.

bootstrap_graph async 

bootstrap_graph(pool: Any, runtime_login: str, *, edge_upsert_mode: str = 'tenant_guarded') -> None

Owner-only opt-in first install/upgrade, then grant existing login DML.

Never creates roles, changes passwords, transfers ownership or swaps indexes. A scoped install requires the operator's scoped index already present. Upgrade is a maintenance-window operation: base ALTER/INDEX locks remain held until commit. A failed check rolls back schema/grants together. No default grants to future tables: each new graph table requires this explicit contract to be revised.

Source code in src/symfonic/memory/backends/postgres_graph_bootstrap.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def bootstrap_graph(
    pool: Any, runtime_login: str, *, edge_upsert_mode: str = "tenant_guarded"
) -> None:
    """Owner-only opt-in first install/upgrade, then grant existing login DML.

    Never creates roles, changes passwords, transfers ownership or swaps indexes.
    A scoped install requires the operator's scoped index already present.
    Upgrade is a maintenance-window operation: base ALTER/INDEX locks remain held
    until commit. A failed check rolls back schema/grants together. No default grants to
    future
    tables: each new graph table requires this explicit contract to be revised.
    """
    if not _ROLE.fullmatch(runtime_login):
        raise ValueError("graph runtime login must be a simple PostgreSQL identifier")
    connection = await pool.acquire()
    try:
        async with connection.transaction():
            role = await connection.fetchrow(
                "SELECT oid, rolcanlogin, rolsuper, rolcreatedb, rolcreaterole, "
                "rolreplication, rolbypassrls FROM pg_roles WHERE rolname=$1",
                runtime_login,
            )
            if (
                not role
                or not role["rolcanlogin"]
                or any(
                    role[key]
                    for key in (
                        "rolsuper",
                        "rolcreatedb",
                        "rolcreaterole",
                        "rolreplication",
                        "rolbypassrls",
                    )
                )
                or await connection.fetchval(
                    "SELECT EXISTS(SELECT 1 FROM pg_auth_members WHERE member=$1)", role["oid"]
                )
            ):
                raise GraphRuntimeNotReady("bootstrap requires an unprivileged existing login")
            if await connection.fetchval(
                "SELECT datdba=$1 FROM pg_database WHERE datname=current_database()", role["oid"]
            ):
                raise GraphRuntimeNotReady("graph runtime must not own the database")

            # Pin the acquired connection: bootstrap DDL and grants share one txn.
            class ConnectionPool:
                async def acquire(self) -> Any:
                    return connection

                async def release(self, acquired: Any) -> None:
                    assert acquired is connection

            await PostgresGraphBackend(
                ConnectionPool(), edge_upsert_mode=edge_upsert_mode
            ).ensure_schema()
            for table in TABLES:
                row = await connection.fetchrow(
                    "SELECT c.relowner=(SELECT oid FROM pg_roles WHERE rolname=current_user) "
                    "AS owns_table, c.relowner=$2 OR n.nspowner=$2 AS runtime_owner, "
                    "has_schema_privilege($2,n.oid,'CREATE') AS runtime_create, "
                    "n.nspname AS schema FROM pg_class c JOIN pg_namespace n "
                    "ON n.oid=c.relnamespace WHERE c.oid=to_regclass($1)",
                    table,
                    role["oid"],
                )
                if (
                    not row
                    or not row["owns_table"]
                    or row["runtime_owner"]
                    or row["runtime_create"]
                ):
                    raise GraphRuntimeNotReady("bootstrap requires separate owner and runtime")
                schema = '"' + row["schema"].replace('"', '""') + '"'
                await connection.execute(f'GRANT USAGE ON SCHEMA {schema} TO "{runtime_login}"')
                await connection.execute(
                    f'GRANT SELECT,INSERT,UPDATE,DELETE ON {schema}.{table} TO "{runtime_login}"'
                )
            try:  # CORE-199 S1: the migration may run before or after bootstrap
                if jobs := await jobs_table(connection):
                    await grant_read_only(connection, jobs, runtime_login)
            except MemoryContractError as error:
                raise GraphRuntimeNotReady(f"bootstrap erase job table: {error}") from error
            from symfonic.memory.backends import postgres_memory_generation as generation

            try:  # CORE-199: the generation migration may run before or after bootstrap
                if generations := await generation.generations_table(connection):
                    await generation.grant_runtime(connection, generations, runtime_login)
            except MemoryContractError as error:
                raise GraphRuntimeNotReady(f"bootstrap erase generation table: {error}") from error
    finally:
        await pool.release(connection)

require_old_login_fenced async 

require_old_login_fenced(connection: Any, old_login: str, *, schema: str) -> None

Read-only cluster check; no termination, credential change or fleet claim.

Use an administrative observer with visibility into pg_stat_activity. NOLOGIN does not stop existing sessions, and prepared transactions survive termination. The caller must separately inventory SECURITY DEFINER functions, other logins and privileged operators. This requires all four tables/schema owned by old_login. Former members with an already active SET ROLE evade this point-in-time probe; the operator must drain their sessions too. Membership history is not observable.

Source code in src/symfonic/memory/backends/postgres_graph_bootstrap.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def require_old_login_fenced(connection: Any, old_login: str, *, schema: str) -> None:
    """Read-only cluster check; no termination, credential change or fleet claim.

    Use an administrative observer with visibility into pg_stat_activity. NOLOGIN
    does not stop existing sessions, and prepared transactions survive termination.
        The caller must separately inventory SECURITY DEFINER functions, other logins
        and privileged operators. This requires all four tables/schema owned by old_login.
        Former members with an already active SET ROLE evade this point-in-time probe;
        the operator must drain their sessions too. Membership history is not observable.
    """
    row = await connection.fetchrow(
        "SELECT oid, rolcanlogin FROM pg_roles WHERE rolname=$1", old_login
    )
    if not row or row["rolcanlogin"]:
        raise GraphRuntimeNotReady("old graph login is not disabled")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_auth_members WHERE roleid=$1)", row["oid"]
    ):
        raise GraphRuntimeNotReady("old graph login still has member roles")
    for table in TABLES:
        if not await connection.fetchval(
            "SELECT c.relowner=$1 AND n.nspowner=$1 FROM pg_class c "
            "JOIN pg_namespace n ON n.oid=c.relnamespace "
            "WHERE c.oid=to_regclass(format('%I.%I',$2::text,$3::text))",
            row["oid"],
            schema,
            table,
        ):
            raise GraphRuntimeNotReady("old graph login does not own the checked graph schema")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE usesysid=$1)", row["oid"]
    ):
        raise GraphRuntimeNotReady("old graph login still has sessions")
    if await connection.fetchval(
        "SELECT EXISTS(SELECT 1 FROM pg_prepared_xacts WHERE owner=$1)", old_login
    ):
        raise GraphRuntimeNotReady("old graph login still has prepared transactions")