Skip to content

symfonic.memory.backends.postgres_staged_fence

postgres_staged_fence

Durable database fence for receipt-claimed pending rows.

A receipt-written row carries its selected claim in the same statement that marks it pending. Until the exact selected publication runs, no UPDATE may take such a row out of pending, change its claim, or move it, whichever code path issues the statement: generic flush, reconciliation, graph-only publish, or a raw backend update_node. The receipt publisher authorizes exactly one claim at a time with a transaction-local setting before its UPDATE. Any session can set that setting, so the fence guards framework code paths, not a hostile writer; postgres_staged_roles is the opt-in database authority. DELETE stays allowed, so privacy erasure is never blocked; an erased staged row simply makes activation refuse.

The trigger is installed by the opt-in staged journal schema. Generic pending selectors exclude claimed rows independently, so they skip instead of failing.

authorize async 

authorize(conn: Any, claim: str) -> None

Admit exactly claim for the rest of the caller's transaction.

Source code in src/symfonic/memory/backends/postgres_staged_fence.py 
109
110
111
112
113
async def authorize(conn: Any, claim: str) -> None:
    """Admit exactly ``claim`` for the rest of the caller's transaction."""
    if type(claim) is not str or not claim:
        raise MemoryContractError("selected publication requires its claim")
    await conn.execute("SELECT set_config($1, $2, true)", SETTING, claim)

claimed_pending 

claimed_pending(node: Any) -> bool

A receipt-claimed row that its selected publication has not released.

Source code in src/symfonic/memory/backends/postgres_staged_fence.py 
71
72
73
def claimed_pending(node: Any) -> bool:
    """A receipt-claimed row that its selected publication has not released."""
    return is_receipt_claimed(getattr(node, "properties", None))

refuse_claimed 

refuse_claimed(nodes: Any) -> None

Generic descendant lifecycle fails closed over any claimed pending row.

Source code in src/symfonic/memory/backends/postgres_staged_fence.py 
90
91
92
93
94
def refuse_claimed(nodes: Any) -> None:
    """Generic descendant lifecycle fails closed over any claimed pending row."""
    if any(claimed_pending(node) for node in nodes):
        raise MemoryContractError(
            "receipt-claimed pending rows require their selected publication")

require_installed async 

require_installed(conn: Any) -> None

Refuse staged work on a database whose fence is absent or disabled.

Source code in src/symfonic/memory/backends/postgres_staged_fence.py 
 97
 98
 99
100
101
102
103
104
105
106
async def require_installed(conn: Any) -> None:
    """Refuse staged work on a database whose fence is absent or disabled."""
    enabled = await conn.fetchval(
        # pgvector registration makes asyncpg decode catalog "char" as bytes.
        "SELECT tgenabled::text FROM pg_trigger WHERE tgname=$1 "
        "AND tgrelid='memory_nodes'::regclass AND NOT tgisinternal "
        "AND tgfoid=to_regprocedure($1 || '()') AND tgtype=19", FUNCTION,
    )
    if enabled not in ("O", "A"):
        raise MemoryContractError("staged publication requires the receipt claim fence")

without_claimed 

without_claimed(name: str, result: Any) -> Any

Hide claimed pending rows from one node-returning read result.

Source code in src/symfonic/memory/backends/postgres_staged_fence.py 
76
77
78
79
80
81
82
83
84
85
86
87
def without_claimed(name: str, result: Any) -> Any:
    """Hide claimed pending rows from one node-returning read result."""
    if name == "get_node":
        return None if result is not None and claimed_pending(result) else result
    if name == "select_candidates":
        from dataclasses import replace

        return replace(result, nodes=tuple(
            node for node in result.nodes if not claimed_pending(node)))
    if name in NODE_READS:
        return [node for node in result if not claimed_pending(node)]
    return result