Skip to content

symfonic.memory.backends.postgres_edge_scoped_writer

postgres_edge_scoped_writer

Explicit, dark PostgreSQL edge writer for the scoped unique-index cutover.

This module never builds or removes indexes. A fleet write barrier and index swap are operator work; a single connection cannot establish fleet state.

ScopedEdgeIndexNotReady

Bases: RuntimeError

The resolved edge table cannot safely serve scoped conflict writes.

ScopedEdgeWriteConflict

Bases: RuntimeError

A different unique constraint rejected the edge write.

require_scoped_edge_index async 

require_scoped_edge_index(conn: Any) -> None

Check the exact table resolved by the writer on this connection.

Source code in src/symfonic/memory/backends/postgres_edge_scoped_writer.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
async def require_scoped_edge_index(conn: Any) -> None:
    """Check the exact table resolved by the writer on this connection."""
    try:
        indexes = await conn.fetch(_INDEX_SQL)
        if (
            not indexes or indexes[0]["relkind"] != "r"
            or indexes[0]["identity_collations_deterministic"] is not True
        ):
            raise ScopedEdgeIndexNotReady("scoped edge index is not ready")
        for index in indexes:
            keys = index["keys"]
            if index["name"] == OLD_INDEX or (
                index["unique"] and index["key_count"] == 4
                and tuple(keys) == _KEYS
            ):
                raise ScopedEdgeIndexNotReady("tenant-wide edge index is still present")
            if (
                index["name"] != NEW_INDEX and index["unique"]
                and index["key_count"] == 5 and tuple(keys[:4]) == _KEYS
            ):
                # ON CONFLICT may infer every matching unique index. An extra
                # expression index can arbitrate with a different collation.
                raise ScopedEdgeIndexNotReady("another scoped edge arbiter is present")
        matches = [index for index in indexes if index["name"] == NEW_INDEX]
        if len(matches) != 1:
            raise ScopedEdgeIndexNotReady("scoped edge index is not ready")
        index = matches[0]
        keys = index["keys"]
        if not (
            index["unique"] and index["valid"] and index["ready"]
            and index["live"] and index["nonpartial"]
            and index["key_collations_deterministic"] is True
            and index["key_opclasses_default"] is True
            and index["method"] == "btree"
            and index["key_count"] == index["total_count"] == 5
            and len(keys) == 5 and tuple(keys[:4]) == _KEYS
            and _canonical_expression(keys[4]) in _EXPRESSIONS
        ):
            raise ScopedEdgeIndexNotReady("scoped edge index is not ready")
    except ScopedEdgeIndexNotReady:
        raise
    except Exception:
        # Driver errors can contain SQL values. Keep refusal independent of them.
        raise ScopedEdgeIndexNotReady("scoped edge index verification failed") from None

upsert_scoped_edge async 

upsert_scoped_edge(conn: Any, scope: TenantScope, edge: MemoryEdge) -> Any

Verify then write through one connection; preserve the existing contract.

Source code in src/symfonic/memory/backends/postgres_edge_scoped_writer.py 
149
150
151
152
153
154
155
156
157
async def upsert_scoped_edge(conn: Any, scope: TenantScope, edge: MemoryEdge) -> Any:
    """Verify then write through one connection; preserve the existing contract."""
    native = _is_native_scope(scope)
    async with conn.transaction() if native else nullcontext():
        if native:
            from symfonic.memory.backends.postgres_native_edge_write import lock_native_endpoints

            await lock_native_endpoints(conn, scope, edge)
        return await _upsert_scoped_edge(conn, scope, edge)