symfonic.memory.backends.postgres_working_guard

postgres_working_guard

Explicit v0 PostgreSQL Working endpoint registration and retirement.

Only a host that already admitted an owner may call these operations. They require the host's transaction on the same pool as its graph backend; the database trigger also fences writers running older Python binaries.

WorkingEndpointGuard dataclass

WorkingEndpointGuard(pool: Any, schema: str, working_host_role: str)

Explicit host binding; never infer a guard schema from search_path.

install_working_endpoint_guard async

install_working_endpoint_guard(pool: Any, *, working_host_role: str) -> None

Install explicitly as an offline migration before Working admission.

This locks the graph tables against concurrent writes while checking that the reserved namespace is unused, then installs the triggers atomically. Rehearse the lock cost on staging; this is not a hot-rollout hook.

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

async def install_working_endpoint_guard(pool: Any, *, working_host_role: str) -> None:
    """Install explicitly as an offline migration before Working admission.

    This locks the graph tables against concurrent writes while checking that
    the reserved namespace is unused, then installs the triggers atomically.
    Rehearse the lock cost on staging; this is not a hot-rollout hook.
    """
    if not re.fullmatch(r"[a-z][a-z0-9_]{0,62}", working_host_role):
        raise ValueError("Working host role must be a simple PostgreSQL identifier")
    if bound_connection(pool) is not None:
        raise RuntimeError("Working guard installation requires its own transaction")
    conn = await pool.acquire()
    try:
        async with conn.transaction():
            await conn.execute("SET LOCAL lock_timeout = '5s'")
            await conn.execute(
                "SELECT pg_advisory_xact_lock("
                "hashtextextended('symfonic-working-guard-install', 0))"
            )
            schema = await conn.fetchval('SELECT current_schema()')
            if schema is None:
                raise RuntimeError('Working guard requires a trusted current schema')
            quoted_schema = '"' + schema.replace('"', '""') + '"'
            installer_role = await conn.fetchval("SELECT current_user")
            role = await conn.fetchrow(
                "SELECT oid, rolsuper, rolcreaterole, rolbypassrls, rolcanlogin "
                "FROM pg_roles WHERE rolname=$1", working_host_role,
            )
            owners = await conn.fetch(
                "SELECT c.relowner FROM pg_class c JOIN pg_namespace n "
                "ON n.oid=c.relnamespace WHERE n.nspname=$1 "
                "AND c.relname=ANY($2::text[])",
                schema, ["memory_nodes", "memory_edges"],
            )
            if (
                role is None or working_host_role == installer_role
                or role["rolsuper"] or role["rolcreaterole"]
                or role["rolbypassrls"] or not role["rolcanlogin"]
                or len(owners) != 2
            ):
                raise RuntimeError("Working guard requires a separate unprivileged host role")
            for row in owners:
                if await conn.fetchval(
                    "SELECT pg_has_role($1::oid,$2::oid,'MEMBER')",
                    role["oid"], row["relowner"],
                ):
                    raise RuntimeError("Working host role cannot inherit table ownership")
            quoted_role = '"' + working_host_role + '"'
            expected_marker = 'symfonic-working-guard-v1:' + working_host_role
            ddl = WORKING_GUARD_DDL.replace('__GUARD_SCHEMA__', quoted_schema)
            marker = await conn.fetchval(
                "SELECT obj_description(to_regclass($1)::oid, 'pg_class')",
                f'{quoted_schema}.memory_working_endpoint_guard',
            )
            if marker is not None:
                if marker != expected_marker:
                    raise RuntimeError('Working guard schema conflicts with existing table')
                await attest_working_guard(
                    conn, schema, quoted_schema, ddl, working_host_role,
                )
                return
            existing = await conn.fetchval(
                "SELECT to_regclass($1) IS NOT NULL",
                f'{quoted_schema}.memory_working_endpoint_guard',
            )
            if existing:
                raise RuntimeError('Working guard table exists without version marker')
            await conn.execute(
                f'LOCK TABLE {quoted_schema}.memory_nodes, {quoted_schema}.memory_edges '
                'IN SHARE ROW EXCLUSIVE MODE'
            )
            collision = await conn.fetchval(
                f'SELECT EXISTS (SELECT 1 FROM {quoted_schema}.memory_nodes '
                'WHERE id COLLATE "C" LIKE $1) OR '
                f'EXISTS (SELECT 1 FROM {quoted_schema}.memory_edges '
                'WHERE source COLLATE "C" LIKE $1 OR target COLLATE "C" LIKE $1)',
                'symfonic-working-v1:%',
            )
            if collision:
                raise RuntimeError('Reserved Working identifier namespace is already in use')
            await conn.execute(ddl)
            await conn.execute(
                f'GRANT USAGE ON SCHEMA {quoted_schema} TO {quoted_role}'
            )
            await conn.execute(
                f'GRANT SELECT, INSERT, UPDATE ON '
                f'{quoted_schema}.memory_working_endpoint_guard TO {quoted_role}'
            )
            await conn.execute(
                f"COMMENT ON TABLE {quoted_schema}.memory_working_endpoint_guard "
                f"IS '{expected_marker}'"
            )
            await attest_working_guard(
                conn, schema, quoted_schema, ddl, working_host_role,
            )
    finally:
        await pool.release(conn)

register_working_endpoint async

register_working_endpoint(guard: WorkingEndpointGuard, scope: TenantScope, record_id: str) -> None

Register a newly created Working node in the same transaction.

Any preexisting incident edge, including a legacy orphan, prevents an old edge from silently becoming attached to this new Working identity.

Advisory contention refuses with SQLSTATE 23514 instead of waiting, even without a lock cycle. Roll back the host transaction before any retry.

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

async def register_working_endpoint(
    guard: WorkingEndpointGuard, scope: TenantScope, record_id: str,
) -> None:
    """Register a newly created Working node in the same transaction.

    Any preexisting incident edge, including a legacy orphan, prevents an old
    edge from silently becoming attached to this new Working identity.

    Advisory contention refuses with SQLSTATE 23514 instead of waiting, even
    without a lock cycle. Roll back the host transaction before any retry.
    """
    conn = await _transaction_connection(guard)
    schema = guard.quoted_schema
    tenant, owner = _v0_owner(scope)
    if not record_id.startswith('symfonic-working-v1:'):
        raise ValueError('Working guard requires a reserved Working identifier')
    isolation = await conn.fetchval("SELECT current_setting('transaction_isolation')")
    if isolation != 'read committed':
        raise RuntimeError('Working endpoint mutation requires READ COMMITTED')
    node = await conn.fetchrow(
        f'SELECT layer, scope_path FROM {schema}.memory_nodes '
        'WHERE tenant_id=$1 AND id=$2 AND '
        'tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 FOR UPDATE',
        tenant, record_id,
    )
    if node is None or node["layer"] != "working" or node["scope_path"] != owner:
        raise ValueError("Working registration requires an exact owned node")
    await _try_advisory_fence(conn, tenant + chr(31) + record_id)
    incident = await conn.fetchval(
        f'SELECT EXISTS (SELECT 1 FROM {schema}.memory_edges '
        'WHERE tenant_id=$1 AND tenant_id COLLATE "C"=$1 '
        'AND (source=$2 OR target=$2) '
        'AND (source COLLATE "C"=$2 OR target COLLATE "C"=$2))',
        tenant, record_id,
    )
    if incident:
        raise ValueError("Working registration refuses preexisting incident edges")
    await conn.execute(
        f'INSERT INTO {schema}.memory_working_endpoint_guard '
        '(tenant_id, record_id, scope_path) VALUES ($1,$2,$3)',
        tenant, record_id, owner,
    )

retire_working_endpoint async

retire_working_endpoint(guard: WorkingEndpointGuard, scope: TenantScope, record_id: str) -> None

Tombstone, remove same-owner edges, then delete the node atomically.

A foreign-owned incident edge is a refusal, never an invitation to delete another scope's data. Tombstones remain so delayed legacy writes cannot recreate an incident edge after the node is gone.

Tenant or endpoint advisory contention refuses with SQLSTATE 23514 instead of waiting, even for unrelated retirements in the same tenant. Roll back the host transaction before any retry. This is a narrow advisory fence; unmanaged savepoints/caught errors and other row locks remain unresolved.

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

async def retire_working_endpoint(
    guard: WorkingEndpointGuard, scope: TenantScope, record_id: str,
) -> None:
    """Tombstone, remove same-owner edges, then delete the node atomically.

    A foreign-owned incident edge is a refusal, never an invitation to delete
    another scope's data. Tombstones remain so delayed legacy writes cannot
    recreate an incident edge after the node is gone.

    Tenant or endpoint advisory contention refuses with SQLSTATE 23514 instead
    of waiting, even for unrelated retirements in the same tenant. Roll back
    the host transaction before any retry. This is a narrow advisory fence;
    unmanaged savepoints/caught errors and other row locks remain unresolved.
    """
    conn = await _transaction_connection(guard)
    schema = guard.quoted_schema
    tenant, owner = _v0_owner(scope)
    if not record_id.startswith('symfonic-working-v1:'):
        raise ValueError('Working guard requires a reserved Working identifier')
    isolation = await conn.fetchval("SELECT current_setting('transaction_isolation')")
    if isolation != 'read committed':
        raise RuntimeError('Working endpoint mutation requires READ COMMITTED')
    await _try_advisory_fence(conn, "symfonic-working-retire:" + tenant)
    node = await conn.fetchrow(
        f'SELECT layer, scope_path FROM {schema}.memory_nodes '
        'WHERE tenant_id=$1 AND id=$2 AND '
        'tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 FOR UPDATE',
        tenant, record_id,
    )
    if node is None or node["layer"] != "working" or node["scope_path"] != owner:
        raise ValueError("Working retirement requires its exact owned node")
    await _try_advisory_fence(conn, tenant + chr(31) + record_id)
    guard = await conn.fetchrow(
        f'SELECT scope_path, retired FROM {schema}.memory_working_endpoint_guard '
        'WHERE tenant_id COLLATE "C"=$1 AND record_id COLLATE "C"=$2 FOR UPDATE',
        tenant, record_id,
    )
    if guard is None or guard["scope_path"] != owner or guard["retired"]:
        raise ValueError("Working retirement requires a live exact owner")
    foreign = await conn.fetchval(
        f'SELECT EXISTS (SELECT 1 FROM {schema}.memory_edges '
        'WHERE tenant_id=$1 AND tenant_id COLLATE "C"=$1 '
        'AND (source=$2 OR target=$2) '
        'AND (source COLLATE "C"=$2 OR target COLLATE "C"=$2) '
        'AND (scope_path IS NULL OR scope_path COLLATE "C"<>$3))',
        tenant, record_id, owner,
    )
    if foreign:
        raise ValueError("Working retirement refuses foreign-owned incident edges")
    await conn.execute(
        f'UPDATE {schema}.memory_working_endpoint_guard SET retired=true '
        'WHERE tenant_id COLLATE "C"=$1 AND record_id COLLATE "C"=$2',
        tenant, record_id,
    )
    await conn.execute(
        f'DELETE FROM {schema}.memory_edges WHERE tenant_id=$1 AND tenant_id COLLATE "C"=$1 '
        'AND (source=$2 OR target=$2) '
        'AND (source COLLATE "C"=$2 OR target COLLATE "C"=$2) '
        'AND scope_path=$3 AND scope_path COLLATE "C"=$3',
        tenant, record_id, owner,
    )
    result = await conn.execute(
        f'DELETE FROM {schema}.memory_nodes WHERE tenant_id=$1 AND id=$2 AND scope_path=$3 '
        'AND tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 '
        'AND scope_path COLLATE "C"=$3',
        tenant, record_id, owner,
    )
    if result != "DELETE 1":
        raise ValueError("Working retirement lost its owned node")