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
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)
|