symfonic.memory.backends.postgres_operation_receipt

postgres_operation_receipt

Opt-in PostgreSQL cycle receipts; the plain pool remains receipt-free.

PostgresReceiptDomain

PostgresReceiptDomain(pool: Any)

Bases: ReceiptSelected, ReceiptWrites

One owned PostgreSQL pool with a durable single-use operation identity.

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

def __init__(self, pool: Any) -> None:
    from symfonic.memory.backends.pool import PostgresPoolManager

    if type(pool) is not PostgresPoolManager:
        raise MemoryContractError("receipt domain requires an exact PostgresPoolManager")
    self._pool = pool
    from symfonic.memory.backends.postgres_graph import PostgresGraphBackend
    from symfonic.memory.backends.postgres_receipt_views import (
        PostgresReceiptGraph,
        PostgresReceiptVector,
    )
    from symfonic.memory.backends.postgres_vector import PostgresVectorBackend

    self.graph = PostgresReceiptGraph(PostgresGraphBackend(pool), self)
    self.vector = PostgresReceiptVector(PostgresVectorBackend(pool), self)
    self._states: WeakKeyDictionary[asyncio.Task[Any], ReceiptState] = WeakKeyDictionary()

pool property

pool: Any

The receipt, graph and vector pool cannot be rebound after construction.

abandon_reservation

abandon_reservation(operation: MemoryOperation, *, owner: str) -> None

Release only local state; the durable PENDING identity remains spent.

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

def abandon_reservation(self, operation: MemoryOperation, *, owner: str) -> None:
    """Release only local state; the durable PENDING identity remains spent."""
    state = self._state()
    if state.active:
        raise MemoryContractError("cannot abandon an active receipt transaction")
    if state.reserved == (operation, owner):
        state.reserved = None

arm_staged_proof

arm_staged_proof(armed: Any) -> None

This task's next transaction for the armed FLUSH writes its proof.

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

def arm_staged_proof(self, armed: Any) -> None:
    """This task's next transaction for the armed FLUSH writes its proof."""
    from symfonic.memory.backends.postgres_staged_proof import require_armed

    state = self._state()
    if state.active or state.staged_proof is not None:
        raise MemoryContractError("staged proof task is busy")
    state.staged_proof = require_armed(armed)

bind_reserved

bind_reserved(operation: MemoryOperation, *, owner: str) -> None

Join a same-transaction cadence reservation in this task only.

The following operation_transaction rechecks the durable row and owner under FOR UPDATE before any publication can commit.

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

def bind_reserved(self, operation: MemoryOperation, *, owner: str) -> None:
    """Join a same-transaction cadence reservation in this task only.

    The following operation_transaction rechecks the durable row and owner
    under FOR UPDATE before any publication can commit.
    """
    require_operation(operation, MemoryOperationVerb.COMMIT_CYCLE, operation.scope)
    state = self._state()
    if (state.active or state.reserved is not None or type(owner) is not str
            or not owner or bound_connection(self.pool) is not None):
        raise MemoryContractError("receipt reservation task is busy")
    state.reserved = (operation, owner)

reconcile async

reconcile(operation: MemoryOperation) -> MemoryOperationOutcome

Only a matching COMMITTED row is positive; never infer ABORTED.

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

async def reconcile(self, operation: MemoryOperation) -> MemoryOperationOutcome:
    """Only a matching COMMITTED row is positive; never infer ABORTED."""
    if type(operation) is not MemoryOperation:
        raise MemoryContractError("reconciliation requires exact MemoryOperation")
    conn = None
    try:
        conn = await self.pool.acquire()
        row = await conn.fetchrow(
            "SELECT * FROM memory_operation_receipts "
            "WHERE tenant_id=$1 AND operation_id=$2",
            operation.scope.tenant, operation.operation_id,
        )
        if row is not None and _exact(row, operation) and row["state"] == "COMMITTED":
            return MemoryOperationOutcome.COMMITTED
    except Exception:
        pass
    finally:
        if conn is not None:
            await self.pool.release(conn)
    return MemoryOperationOutcome.UNKNOWN

reserve async

reserve(operation: MemoryOperation, *, owner: str) -> bool

Spend an ID before heartbeat/phases; only exact COMMITTED may skip.

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

async def reserve(self, operation: MemoryOperation, *, owner: str) -> bool:
    """Spend an ID before heartbeat/phases; only exact COMMITTED may skip."""
    require_operation(operation, operation.verb, operation.scope)
    state = self._state()
    if state.reserved is not None and state.reserved[0] != operation:
        raise MemoryContractError("receipt task is busy with another operation")
    if type(owner) is not str or not owner or state.active:
        raise MemoryContractError("receipt reservation requires a lease owner")
    if bound_connection(self.pool) is not None:
        raise MemoryContractError("receipt reservation must commit independently")
    conn = await self.pool.acquire()
    try:
        inserted = await conn.fetchval(RESERVE, *_identity(operation), owner)
        if inserted:
            state.reserved = (operation, owner)
            return True
        row = await conn.fetchrow(
            "SELECT * FROM memory_operation_receipts "
            "WHERE tenant_id=$1 AND operation_id=$2",
            operation.scope.tenant, operation.operation_id,
        )
        if row is not None and _exact(row, operation) and row["state"] == "COMMITTED":
            return False
        raise MemoryOperationOutcomeUnknown(operation)
    finally:
        await self.pool.release(conn)

PostgresReceiptLeases

PostgresReceiptLeases(domain)

Claims and the live lease fate-share one pool transaction.

Owner tokens cannot be reused within the same full scope, even after release. Legacy claimless leases can fence but refuse renewal/release and recovery.

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

def __init__(self, domain) -> None:
    from symfonic.memory.backends.postgres_operation_receipt import PostgresReceiptDomain

    if type(domain) is not PostgresReceiptDomain:
        raise MemoryContractError("receipt leases require exact domain")
    self.transaction_domain = domain
    self._leases = PostgresLeases(domain.pool)

committed_on async

committed_on(conn: Any, operation: MemoryOperation) -> bool

Exact COMMITTED receipt, read and share-locked on the caller's connection.

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

async def committed_on(conn: Any, operation: MemoryOperation) -> bool:
    """Exact COMMITTED receipt, read and share-locked on the caller's connection."""
    row = await conn.fetchrow(
        "SELECT * FROM memory_operation_receipts "
        "WHERE tenant_id=$1 AND operation_id=$2 FOR SHARE",
        operation.scope.tenant, operation.operation_id,
    )
    return row is not None and _exact(row, operation) and row["state"] == "COMMITTED"