symfonic.memory.backends.postgres_claimed_discard

postgres_claimed_discard

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

Reproduces _native_cascade(claim_token=...) applied row by row in the given order. Nodes and their incident edges are locked in chunked statements; the per-node checks then run in Python in the original order, so a single failing check raises the same error. When several coincide, such as a lock wait and a budget overflow, another may surface first; nothing commits either way. An edge is deleted iff the row-at-a-time cascade would have deleted it: the earliest discarded endpoint whose owner rule admits it takes it, and later endpoints no longer see it, either for eligibility or for their incident-edge budget. A node whose locked page was truncated after an earlier deletion re-reads exactly the page the sequential query would see. The whole transaction runs as index probes, and the claim-generation delete requires the operator-installed record index, so no statement reads history.

discard_claimed async

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

Discard (storage scope, node id, token) of one tenant, or none of them.

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

async def discard_claimed(
    pool: Any, tenant_id: str, owned: list[tuple[Any, str, str]], edge_upsert_mode: str,
) -> None:
    """Discard ``(storage scope, node id, token)`` of one tenant, or none of them."""
    conn = await pool.acquire()
    try:
        async with conn.transaction(), index_probes(conn):
            await require_claim_record_index(conn)
            locked = await _lock_nodes(conn, tenant_id, [node_id for _, node_id, _ in owned])
            await refuse_if_erasing(conn, tenant_id, [scope for scope, _, _ in owned])
            doomed = await _doomed_edges(conn, tenant_id, owned, locked, edge_upsert_mode)
            for chunk in _chunks(doomed):
                await conn.execute(EDGES_DELETE_SQL, tenant_id, chunk)
            for chunk in _chunks(owned):
                await _delete_chunk(conn, tenant_id, chunk, locked)
            # Every claim generation of a discarded row goes, as row by row:
            # one record-index descent per row plus the generations it removes.
            await conn.execute(
                CLAIMS_DELETE_SQL, tenant_id, [node_id for _, node_id, _ in owned],
                [locked[node_id]["scope_path"] for _, node_id, _ in owned],
            )
    finally:
        await pool.release(conn)