symfonic.memory.backends.postgres_receipt_selected

postgres_receipt_selected

Selected-publication checks for the opt-in PostgreSQL receipt domain.

ReceiptSelected

SourceWriteRefused

Bases: MemoryContractError

The accepted source WRITE is absent, inexact or no longer this manifest.

Permanent: a retry re-reads the same durable receipt and refuses again.

bound_write_selected

bound_write_selected(domain: Any, flush: MemoryOperation, write: MemoryOperation)

Bind this task's next selected publication to its source WRITE.

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

@contextmanager
def bound_write_selected(domain: Any, flush: MemoryOperation, write: MemoryOperation):
    """Bind this task's next selected publication to its source WRITE."""
    task = asyncio.current_task()
    if task is None or _SELECTED_WRITE.get() is not None:
        raise MemoryContractError("selected WRITE source binding is busy")
    token = _SELECTED_WRITE.set((domain, task, flush, write))
    try:
        yield
    finally:
        _SELECTED_WRITE.reset(token)

committed_manifest_on async

committed_manifest_on(conn: Any, write: MemoryOperation) -> Any

The source WRITE's COMMITTED manifest, share-locked on conn.

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

async def committed_manifest_on(conn: Any, write: MemoryOperation) -> Any:
    """The source WRITE's COMMITTED manifest, share-locked on ``conn``."""
    from symfonic.capabilities.memory.selected_records import SelectedRecordManifest
    from symfonic.memory.backends.postgres_operation_receipt import _exact

    row = await conn.fetchrow(
        "SELECT * FROM memory_operation_receipts "
        "WHERE tenant_id=$1 AND operation_id=$2 FOR SHARE",
        write.scope.tenant, write.operation_id,
    )
    if (row is None or not _exact(row, write) or row["state"] != "COMMITTED"
            or row["write_manifest"] is None):
        return None
    document = row["write_manifest"]
    try:
        manifest = SelectedRecordManifest.from_document(
            json.loads(document) if isinstance(document, str) else document,
        )
        manifest.require_scope(write.scope)
    except MemoryContractError:
        return None  # a rewritten manifest is not this source; never retryable
    if any(claim.owner != write.scope for claim in manifest.claims):
        return None
    return manifest

operation_for_selected_write

operation_for_selected_write(operation_id: str, write: MemoryOperation, manifest: Any) -> MemoryOperation

Canonical FLUSH of exactly one accepted WRITE's durable manifest.

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

def operation_for_selected_write(
    operation_id: str, write: MemoryOperation, manifest: Any,
) -> MemoryOperation:
    """Canonical FLUSH of exactly one accepted WRITE's durable manifest."""
    if type(write) is not MemoryOperation or write.verb is not MemoryOperationVerb.WRITE:
        raise MemoryContractError("selected FLUSH requires an exact source WRITE identity")
    return MemoryOperation(operation_id, MemoryOperationVerb.FLUSH, write.scope, {
        **manifest.intent(),
        "accepted_write": {"operation_id": write.operation_id,
                           "intent_digest": write.intent_digest},
    })

require_selected_write

require_selected_write(operation: Any, write: Any, manifest: Any) -> None

Refuse any FLUSH other than the canonical one for this durable manifest.

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

def require_selected_write(operation: Any, write: Any, manifest: Any) -> None:
    """Refuse any FLUSH other than the canonical one for this durable manifest."""
    if (
        manifest is None
        or type(operation) is not MemoryOperation
        or operation_for_selected_write(operation.operation_id, write, manifest) != operation
        or operation.selected_manifest != manifest
    ):
        raise SourceWriteRefused("selected FLUSH differs from its accepted WRITE manifest")