symfonic.memory.backends.postgres_working_transaction

postgres_working_transaction

Bounded replay of a host's complete PostgreSQL Working write.

WorkingTransactionUncertain

Bases: RuntimeError

Control/cleanup acknowledgement missing; never proof of rollback.

run_working_read async

run_working_read(pool, read, *, deadline_at)

Own a bounded read-only readback; no retry and no caller transaction.

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

async def run_working_read(pool, read, *, deadline_at):
    """Own a bounded read-only readback; no retry and no caller transaction."""
    if bound_connection(pool) is not None:
        raise RuntimeError("Working readback refuses ambient transaction")
    return await _with_deadline(pool, read, deadline_at, read_only=True)

run_working_transaction async

run_working_transaction(pool: Any, write: Callable[[Any], Awaitable[_Result]], *, deadline_at: float | None = None) -> _Result

Replay only a deadlocked body after confirmed whole-transaction rollback.

write must run every database statement on conn (directly or through this exact pool's task-bound connection), never another pool or connection. It must contain the complete database write and no external effects, audit publication, transaction control, or committed-state readback. It may run three times. Ambient transactions are refused: a savepoint cannot replay earlier owner admission or release the caller's earlier locks. Commit and rollback errors propagate without replay, including uncertain outcomes.

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

async def run_working_transaction(
    pool: Any, write: Callable[[Any], Awaitable[_Result]], *, deadline_at: float | None = None,
) -> _Result:
    """Replay only a deadlocked body after confirmed whole-transaction rollback.

    ``write`` must run every database statement on ``conn`` (directly or through
    this exact pool's task-bound connection), never another pool or connection.
    It must contain the complete database write and no external effects,
    audit publication, transaction control, or committed-state readback. It may
    run three times. Ambient transactions are refused: a savepoint cannot replay
    earlier owner admission or release the caller's earlier locks. Commit and
    rollback errors propagate without replay, including uncertain outcomes.
    """
    if bound_connection(pool) is not None:
        raise RuntimeError("Working retry requires its own PostgreSQL transaction")
    if deadline_at is not None:
        return await _with_deadline(pool, write, deadline_at)
    for attempt in range(_ATTEMPTS):
        deadlock = None
        try:
            async with run_in_transaction(pool) as conn:
                try:
                    result = await write(conn)
                except Exception as exc:
                    if getattr(exc, "sqlstate", None) == "40P01" and conn.is_in_transaction():
                        deadlock = exc
                    raise
                if not conn.is_in_transaction():
                    raise RuntimeError("Working write callback ended its transaction")
            return result
        except Exception as exc:
            # Context-manager exit completed rollback before this comparison.
            if exc is not deadlock or attempt == _ATTEMPTS - 1:
                raise
    raise AssertionError("Working retry exhausted without an outcome")