symfonic.memory.backends.postgres_staged_content

postgres_staged_content

Per-record content seal for one staged PostgreSQL v1 turn (CORE-339a).

READY binds the run and the output digest; the claim fence keeps each staged row pending and claimed. Neither binds what a row says: a session that may UPDATE a claimed pending row (the owner, or the CORE-376 writer by policy) can rewrite its label or properties while keeping the claim keys, and a session that may DELETE can remove it. This module closes both:

• seal_records runs inside the READY transaction. For every claim of the sealed FLUSH manifest, in manifest order, it reads the exact row (tenant, id) and records a server-computed SHA-256 over every published column. The result travels inside the AEAD-sealed READY payload, which the writer may append but never rewrite.
• spend runs inside the receipt transaction that publishes the FLUSH. It inserts the FLUSH receipt PENDING there, not in a prior committed transaction, then locks every sealed row FOR UPDATE and requires each to be present with an equal digest, all before any vector is written. A missing or changed row raises, the transaction rolls back, and no receipt, proof, release or vector survives: the turn stays READY rather than UNKNOWN.

The digest is computed by PostgreSQL on both sides, over jsonb text, so it does not depend on client decoding; both sides run inside a transaction with extra_float_digits pinned to 1, the server default, so the float8 importance text does not depend on the session either. created_at, updated_at and access_count are bookkeeping, not published content, and are excluded.

ContentSeal

Bases: NamedTuple

The verified per-record digests and the READY row bytes they came from.

StagedContentRefused

Bases: MemoryContractError

A sealed staged row is absent or no longer equals its READY digest.

require_seal

require_seal(value: Any, sealed_sha256: Any, operation: Any) -> ContentSeal

Exact shape and version, and exactly the manifest's rows in its order.

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

def require_seal(value: Any, sealed_sha256: Any, operation: Any) -> ContentSeal:
    """Exact shape and version, and exactly the manifest's rows in its order."""
    if (type(value) is not dict or set(value) != {"version", "records"}
            or value["version"] != VERSION or type(value["version"]) is not int
            or type(value["records"]) is not list):
        _refuse("READY carries no content seal of a supported version")
    records = []
    for item in value["records"]:
        if (type(item) is not list or len(item) != 3
                or any(type(part) is not str for part in item)
                or not _DIGEST.fullmatch(item[2])):
            _refuse("READY content seal is malformed")
        records.append(tuple(item))
    if [record[:2] for record in records] != [key[:2] for key in _keys(operation)]:
        _refuse("READY content seal does not cover exactly the sealed manifest")
    if type(sealed_sha256) is not str or not _DIGEST.fullmatch(sealed_sha256):
        _refuse("READY row digest is malformed")
    return ContentSeal(tuple(records), sealed_sha256)

seal_records async

seal_records(conn: Any, operation: Any) -> dict[str, Any]

Inside READY: every claimed row present, pending, and digested.

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

async def seal_records(conn: Any, operation: Any) -> dict[str, Any]:
    """Inside READY: every claimed row present, pending, and digested."""
    keys = _keys(operation)
    found = await _fetch(conn, keys, _ROWS)
    records = []
    for tenant, record_id, claim in keys:
        row = found.get((tenant, record_id))
        if row is None or row["pending"] is not True or row["claim"] != claim:
            _refuse("a manifest row is absent, published or claimed elsewhere at READY")
        records.append([tenant, record_id, row["digest"]])
    return {"version": VERSION, "records": records}

spend async

spend(conn: Any, operation: Any, owner: str, seal: ContentSeal) -> None

Inside the receipt transaction: reserve the FLUSH, then prove every row.

A concurrent activation holding the same id blocks this INSERT until it ends; if it committed, nothing is inserted and this attempt is UNKNOWN to its caller, whose fresh snapshot then reads the winner's ACTIVE. The reservation and the row locks wait at most SPEND_LOCK_TIMEOUT, so an activator parked inside its publication, or a session holding a staged row, cannot hang this one: it rolls back with the turn READY. The caller's own lock_timeout is restored for the rest of the transaction.

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

async def spend(conn: Any, operation: Any, owner: str, seal: ContentSeal) -> None:
    """Inside the receipt transaction: reserve the FLUSH, then prove every row.

    A concurrent activation holding the same id blocks this INSERT until it
    ends; if it committed, nothing is inserted and this attempt is UNKNOWN to
    its caller, whose fresh snapshot then reads the winner's ACTIVE. The
    reservation and the row locks wait at most ``SPEND_LOCK_TIMEOUT``, so an
    activator parked inside its publication, or a session holding a staged
    row, cannot hang this one: it rolls back with the turn READY. The caller's
    own ``lock_timeout`` is restored for the rest of the transaction.
    """
    from symfonic.memory.backends.postgres_operation_receipt import RESERVE, _identity

    if type(seal) is not ContentSeal:
        _refuse("activation requires a verified content seal")
    previous = await conn.fetchval("SELECT current_setting('lock_timeout')")
    await conn.execute(f"SET LOCAL lock_timeout = '{SPEND_LOCK_TIMEOUT}'")
    if await conn.fetchval(RESERVE, *_identity(operation), owner) != 1:
        raise MemoryOperationOutcomeUnknown(operation)
    keys = _keys(operation)
    if [record[:2] for record in seal.records] != [key[:2] for key in keys]:
        _refuse("the content seal does not cover exactly the armed manifest")
    found = await _fetch(conn, keys, _LOCKED)
    for tenant, record_id, digest in seal.records:
        row = found.get((tenant, record_id))
        if row is None:
            _refuse(f"staged row {record_id} is absent")
        if row["digest"] != digest:
            _refuse(f"staged row {record_id} changed after READY")
    await conn.execute("SELECT set_config('lock_timeout', $1, true)", previous)