Skip to content

symfonic.memory.backends.postgres_claimed_publish

postgres_claimed_publish

Set-based publication of claimed native pending rows, inside one transaction.

Each chunk applies the exact publish_node CAS to every row in a single statement: the same token, owner and pending predicates, the same published properties computed in Python, and the same claim transition. Any row the CAS does not match makes the counts disagree, and the caller's transaction rolls back. The lifecycle deadline still bounds the whole operation; no chunk ever commits on its own.

publish_claimed async 

publish_claimed(pool: Any, tenant_id: str, owned: list[tuple[Any, Any]]) -> None

Publish (storage scope, node) pairs of one tenant, or refuse them all.

Source code in src/symfonic/memory/backends/postgres_claimed_publish.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def publish_claimed(pool: Any, tenant_id: str, owned: list[tuple[Any, Any]]) -> None:
    """Publish ``(storage scope, node)`` pairs of one tenant, or refuse them all."""
    items = [_item(scope, node) for scope, node in owned]
    size = _subtree.NATIVE_SUBTREE_BUDGET
    conn = await pool.acquire()
    try:
        async with conn.transaction(), index_probes(conn):
            for start in range(0, len(items), size):
                chunk = items[start:start + size]
                row = await conn.fetchrow(
                    _PUBLISH_SQL, tenant_id, json.dumps(dict(chunk)),
                    PENDING_KEY, PENDING_BATCH_KEY,
                )
                if row["updated"] != len(chunk) or row["marked"] != len(chunk):
                    raise MemoryContractError(
                        "native pending claim changed before publication"
                    )
            # Every chunk's node update holds ROW EXCLUSIVE until commit.
            await refuse_if_erasing(conn, tenant_id, [scope for scope, _ in owned])
    finally:
        await pool.release(conn)