symfonic.memory.backends.pool_transaction

pool_transaction

Making several backends commit together without telling any of them.

A consolidation cycle applies its whole batch of graph mutations at once, and has to verify it still holds the scope's lease in the same transaction -- otherwise the check and the write are two moments with a gap between them, and the gap is exactly where another worker takes over.

The obvious way to do that is to pass a connection down: add_node(scope, node, connection=conn). It is also the wrong way. GraphBackend is a protocol with a dozen implementations and none of them should carry a transport detail in its signature to serve one caller; the lease port is a separate protocol that would need the same parameter; and every future backend author would have to thread it through or silently fall out of the transaction.

So the connection is ambient instead. A block opens a transaction, binds its connection here, and every acquire() on that same pool hands the connection back rather than taking a new one. Backends are unchanged and unaware. The binding is task-local, so an ordinary turn running beside the block keeps its own connections; and it is keyed by pool identity, so a deployment with two pools does not accidentally splice them into one transaction.

bound_connection

bound_connection(pool: Any) -> Any | None

pool's transaction connection on this task, or None.

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

def bound_connection(pool: Any) -> Any | None:
    """``pool``'s transaction connection on this task, or ``None``."""
    bound = _TRANSACTION.get()
    return bound[1] if bound is not None and bound[0] is pool else None

is_bound

is_bound(conn: Any) -> bool

Whether conn belongs to a transaction and must not be released.

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

def is_bound(conn: Any) -> bool:
    """Whether ``conn`` belongs to a transaction and must not be released."""
    bound = _TRANSACTION.get()
    return bound is not None and bound[1] is conn

run_in_transaction async

run_in_transaction(pool: Any) -> AsyncIterator[Any]

Run a block so every operation on pool commits or rolls back together.

Nested use joins the outer transaction rather than opening a savepoint: the callers here want "all of this together", and a nested block that could commit on its own would be a quieter way of not having a transaction.

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

@asynccontextmanager
async def run_in_transaction(pool: Any) -> AsyncIterator[Any]:
    """Run a block so every operation on ``pool`` commits or rolls back together.

    Nested use joins the outer transaction rather than opening a savepoint:
    the callers here want "all of this together", and a nested block that could
    commit on its own would be a quieter way of not having a transaction.
    """
    existing = bound_connection(pool)
    if existing is not None:
        yield existing
        return
    conn = await pool.acquire()
    token = _TRANSACTION.set((pool, conn))
    try:
        async with conn.transaction():
            yield conn
    finally:
        _TRANSACTION.reset(token)
        await pool.raw_pool.release(conn)