Skip to content

symfonic.memory.backends.postgres_conversation_anchor

postgres_conversation_anchor

A standalone conversation database's erase anchor, kept outside it (CORE-171).

A PostgresConversationLog composed beside raw PostgreSQL v1 memory is proved by memory's anchor (eraser.guarding). One used on its own has no memory lineage, so a physical copy from before a transcript forget was indistinguishable from the live database and served the erased transcript. Since 11.3.0 such a log carries its own anchor: PostgresConversationLog(pool, erase_anchor=PostgresEraseAnchor(anchor_pool, memory="<unique name>"), receipt_key=key), or refuses every read and write (UNANCHORED_LOG) once its erase generations are installed.

The anchor row lives in the same memory_erase_anchor table and keyspace as memory lineages, keyed (name, schema); the name must be unique to this conversation database. Genesis refuses any name that already holds a row, whatever its schema or kind, so a lineage is never overwritten or shared. The digest is HMAC-SHA256 over a conversation domain tag, the seq and the chained log digest (postgres_conversation_erase_log), so a memory digest never verifies as a conversation one. The checks mirror EraseProof: holder (system_identifier/oid/name; MOVED elsewhere), a different cluster (SAME_CLUSTER), the anchored row present in this database's log with the sealed digest (BEHIND), and NOT_ANCHORED when the acknowledgement after a committed forget fails. The validated logical restore is postgres_conversation_restore.

ConversationAnchorError

Bases: MemoryContractError

The conversation database cannot be proven to hold every acknowledged forget.

ConversationAnchorRequired

Bases: MemoryContractError

A standalone conversation log composed without the proof 11.3.0 requires.

ConversationEraseProof

ConversationEraseProof(anchor: Any, *, receipt_key: Any)

One conversation database's anchor, keyed by the operator's unique name.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
def __init__(self, anchor: Any, *, receipt_key: Any) -> None:
    if type(anchor) is not PostgresEraseAnchor:
        raise TypeError("erase_anchor must be a PostgresEraseAnchor (CORE-171)")
    if type(receipt_key) is not bytes or len(receipt_key) < 32:
        raise ConversationAnchorError(NO_KEY)
    self._anchor, self._secret = anchor, receipt_key

acknowledge_on async

acknowledge_on(conn: Any) -> int

Anchor conn's committed log head in-lineage, or raise NOT_ANCHORED.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
async def acknowledge_on(self, conn: Any) -> int:
    """Anchor ``conn``'s committed log head in-lineage, or raise ``NOT_ANCHORED``."""
    try:
        seq, chained = await erase_log.head(conn)

        async def work(anchor_conn: Any) -> None:
            row = await self.lineage(conn, anchor_conn, lock=True)
            if int(row["seq"]) < seq:
                await anchor_conn.execute(
                    f"UPDATE {ANCHOR_TABLE} SET seq=$3, digest=$4 WHERE memory_system=$1 "
                    "AND memory_schema=$2 AND seq=$5", self._anchor.memory,
                    await conn.fetchval(_BOUND), seq, self.seal(seq, chained), row["seq"])

        await self.in_anchor(work)
    except Exception as error:  # noqa: BLE001 - the erase committed; it is not acked
        raise ConversationAnchorError(NOT_ANCHORED) from error
    return seq

genesis async

genesis(owner: Any) -> None

Owner operation, once per name: anchor log row 0 to this database.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
async def genesis(self, owner: Any) -> None:
    """Owner operation, once per name: anchor log row 0 to this database."""
    seq, chained = await erase_log.head(owner)

    async def work(anchor_conn: Any) -> None:
        holder, schema = await self.identity(owner, anchor_conn)
        await anchor_conn.execute(f"LOCK TABLE {ANCHOR_TABLE} IN SHARE ROW EXCLUSIVE MODE")
        if await anchor_conn.fetchval(_NAMED, self._anchor.memory):
            raise ConversationAnchorError(NAME_TAKEN)
        if seq != 0:
            raise ConversationAnchorError(GENESIS_REFUSED)
        await anchor_conn.execute(f"INSERT INTO {ANCHOR_TABLE} VALUES ($1, $2, $3, 0, $4)",
                                  self._anchor.memory, schema, holder, self.seal(0, chained))

    await self.in_anchor(work)

lineage async

lineage(conn: Any, anchor_conn: Any, *, lock: bool) -> Any

The anchor row once conn holds it and its lineage; else refuse typed.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
async def lineage(self, conn: Any, anchor_conn: Any, *, lock: bool) -> Any:
    """The anchor row once ``conn`` holds it and its lineage; else refuse typed."""
    holder, schema = await self.identity(conn, anchor_conn)
    row = await anchor_conn.fetchrow(ROW + (" FOR UPDATE" if lock else ""),
                                     self._anchor.memory, schema)
    if row is None:
        named = await anchor_conn.fetchval(_NAMED, self._anchor.memory)
        raise ConversationAnchorError(MISBOUND if named else UNANCHORED)
    if row["holder"] != holder:
        raise ConversationAnchorError(MOVED)
    chained = await erase_log.digest_at(conn, int(row["seq"]))
    if chained is None or not hmac.compare_digest(self.seal(int(row["seq"]), chained),
                                                  row["digest"]):
        raise ConversationAnchorError(BEHIND)
    return row

verify async

verify(conn: Any) -> int

Refuse unless conn's database holds the anchor; the anchored seq.

Only a database with neither the log nor any anchor row under this name (a fresh install) is admitted unproven; after genesis a missing log refuses UNLOGGED.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
async def verify(self, conn: Any) -> int:
    """Refuse unless ``conn``'s database holds the anchor; the anchored seq.

    Only a database with neither the log nor any anchor row under this name
    (a fresh install) is admitted unproven; after genesis a missing log
    refuses ``UNLOGGED``.
    """
    from symfonic.memory.backends.postgres_conversation_erase import refuse_if_shadowed

    await refuse_if_shadowed(conn)
    logged = await erase_log.present(conn)
    if logged:
        await erase_log.attest(conn)

    async def work(anchor_conn: Any) -> int:
        if not logged:
            holder, _ = await self.identity(conn, anchor_conn)
            held = await anchor_conn.fetch(_HELD, self._anchor.memory)
            if held:  # a name another database holds, or this one's log was dropped
                mine = any(row["holder"] == holder for row in held)
                raise ConversationAnchorError(erase_log.UNLOGGED if mine else MOVED)
            return 0
        return int((await self.lineage(conn, anchor_conn, lock=False))["seq"])

    seq = await self.in_anchor(work)
    if logged and await erase_log.sealed(conn):
        raise ConversationAnchorError(erase_log.SEALED)
    return seq

ConversationGuard

ConversationGuard(proof: ConversationEraseProof, *, timeout: float = VERIFY_SECONDS)

Hand out a conversation connection only once its anchor verifies on it.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
def __init__(self, proof: ConversationEraseProof, *, timeout: float = VERIFY_SECONDS) -> None:
    self._proof, self._timeout = proof, timeout

anchored

anchored(pool: Any, anchor: Any, receipt_key: Any) -> ConversationEraseProof | None

The log's proof, its guard installed on pool; None for an unanchored log.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
def anchored(pool: Any, anchor: Any, receipt_key: Any) -> ConversationEraseProof | None:
    """The log's proof, its guard installed on ``pool``; None for an unanchored log."""
    from symfonic.memory.backends.postgres_erase_incarnation import install

    if anchor is None:
        if receipt_key is not None:
            raise ConversationAnchorRequired(UNANCHORED_LOG)
        return None
    proof = ConversationEraseProof(anchor, receipt_key=receipt_key)
    install(pool, ConversationGuard(proof))
    return proof

require_proven

require_proven(pool: Any) -> None

Refuse every checkout of pool until a proving guard covers it.

Source code in src/symfonic/memory/backends/postgres_conversation_anchor.py
def require_proven(pool: Any) -> None:
    """Refuse every checkout of ``pool`` until a proving guard covers it."""
    guard = getattr(pool, "guard_connections", None)
    if guard is None:
        raise ConversationAnchorRequired(UNANCHORED_LOG)
    guard(UNPROVEN)