symfonic.memory.backends.postgres_erase_anchor

postgres_erase_anchor

The erase anchor: the memory database's erase-log head, kept outside it (CORE-171).

A physical copy, PITR or promoted replica of the memory cluster is byte-identical to it at the copy point, so only state the copy did not carry can show that it lacks an acknowledged forget. That state is one row per memory database in an operator-provided PostgreSQL database in another cluster: (memory_system, memory_schema) -> (holder, seq, digest). The key is the operator's stable, unique name for the memory database (PostgresEraseAnchor(pool, memory=...), required: two databases sharing an anchor store and a schema name stay distinct lineages) and its schema; holder is the system_identifier, database OID and name serving it; digest is HMAC-SHA256(receipt key, seq || chained log digest). The anchor never holds a readable prefix, and the receipt key is never printed. A database whose identity is not the holder refuses MOVED: a logical restore elsewhere stays closed until rebind (the validated restore) moves the holder, once, at the export's exact seq; the sealed source then refuses too. A physical copy keeps the holder, so the lineage check decides.

verify proves the memory database contains the anchored lineage: its log row anchor.seq exists and its chained digest matches. Every memory connection checkout (postgres_erase_incarnation) and every forget before its delete run it. acknowledge runs after an erase commits: in one anchor transaction it locks the row, re-verifies the lineage and moves the anchor to the local head. A forget returns its receipt only after that; otherwise it raises NOT_ANCHORED: committed, not acknowledged. There is no equal-seq no-op: an anchor at or past the head acknowledges only when the lineage check passed.

Operator duties, not provable here: the anchor lives in a different cluster (verify refuses the same system_identifier), is never restored from a backup, and its loss closes memory. Receipt-key rotation has no supported procedure and refuses (a digest under another key reads as a foreign lineage).

EraseAnchorError

Bases: MemoryContractError

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

EraseProof

EraseProof(anchor: Any, *, schema: str, receipt_key: bytes)

One memory schema's anchor, keyed with the eraser's receipt key.

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

def __init__(self, anchor: Any, *, schema: str, receipt_key: bytes) -> None:
    if type(anchor) is not PostgresEraseAnchor:
        raise TypeError("erase_anchor must be a PostgresEraseAnchor (CORE-171)")
    self._anchor = anchor
    self._key = (anchor.memory, schema)
    self._secret = receipt_key
    self.own_pool: Any = None  # the eraser's own pool: its proofs are taken pinned

acknowledge async

acknowledge(memory_pool: Any) -> int

Anchor the committed local head in-lineage, or raise NOT_ANCHORED.

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

async def acknowledge(self, memory_pool: Any) -> int:
    """Anchor the committed local head in-lineage, or raise ``NOT_ANCHORED``."""
    from symfonic.memory.backends.postgres_memory_erase_log import head

    try:
        memory = await memory_pool.acquire()
        try:
            pinned = self._key[1] if memory_pool is self.own_pool else None
            async with _resolving(memory, pinned):
                seq, chained = await head(memory)

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

                await self._in_anchor(work)
        finally:
            await memory_pool.release(memory)
    except Exception as error:  # noqa: BLE001 - the erase committed; it is not acked
        raise EraseAnchorError(NOT_ANCHORED) from error
    return seq

genesis async

genesis(memory: Any) -> None

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

Refused over a non-empty log or any existing row for the key, whoever holds it: an anchor lost after genesis has no in-database re-genesis, and a restore can never start a fresh lineage under an anchored name.

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

async def genesis(self, memory: Any) -> None:
    """Owner operation, once per anchor key: anchor log row 0 to this database.

    Refused over a non-empty log or any existing row for the key, whoever
    holds it: an anchor lost after genesis has no in-database re-genesis,
    and a restore can never start a fresh lineage under an anchored name.
    """
    from symfonic.memory.backends.postgres_memory_erase_log import head

    seq, chained = await head(memory)

    async def work(conn: Any) -> None:
        holder, row = await self._held(memory, conn, lock=True)
        if seq != 0 or row is not None:
            raise EraseAnchorError(GENESIS_REFUSED)
        await conn.execute(f"INSERT INTO {ANCHOR_TABLE} VALUES ($1, $2, $3, 0, $4)",
                           *self._key, holder, self.seal(0, chained))

    await self._in_anchor(work)

holder async

holder(memory: Any) -> str

This database's anchor identity: system_identifier/oid/name.

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

async def holder(self, memory: Any) -> str:
    """This database's anchor identity: ``system_identifier/oid/name``."""
    return str(await memory.fetchval(_HOLDER))

rebind async

rebind(memory: Any, *, source: str, seq: int) -> None

The validated restore: move the anchor from source to memory, once.

Only when the anchor still names source at exactly seq (the sealed export's head, nothing acknowledged since) and memory's log row seq carries the anchored digest. Afterwards source and any other copy refuse MOVED, and a second use of the export elsewhere refuses; re-running it on the same database (a failed commit) is idempotent.

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

async def rebind(self, memory: Any, *, source: str, seq: int) -> None:
    """The validated restore: move the anchor from ``source`` to ``memory``, once.

    Only when the anchor still names ``source`` at exactly ``seq`` (the
    sealed export's head, nothing acknowledged since) and ``memory``'s log
    row ``seq`` carries the anchored digest. Afterwards ``source`` and any
    other copy refuse ``MOVED``, and a second use of the export elsewhere
    refuses; re-running it on the same database (a failed commit) is idempotent.
    """
    from symfonic.memory.backends.postgres_memory_erase_log import digest_at

    async def work(conn: Any) -> None:
        holder, row = await self._held(memory, conn, lock=True)
        if row is None:
            raise EraseAnchorError(UNANCHORED)
        if row["holder"] not in (source, holder) or int(row["seq"]) != seq:
            raise EraseAnchorError(STALE_EXPORT)
        chained = await digest_at(memory, seq)
        if chained is None or not hmac.compare_digest(self.seal(seq, chained), row["digest"]):
            raise EraseAnchorError(BEHIND)
        if holder != source:
            await conn.execute(
                f"UPDATE {ANCHOR_TABLE} SET holder=$3 WHERE memory_system=$1 "
                "AND memory_schema=$2 AND holder=$4 AND seq=$5",
                *self._key, holder, source, seq)

    await self._in_anchor(work)

seal

seal(seq: int, chained: bytes) -> str

The anchor's keyed digest of log row seq; never printed.

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

def seal(self, seq: int, chained: bytes) -> str:
    """The anchor's keyed digest of log row ``seq``; never printed."""
    message = seq.to_bytes(8, "big") + chained
    return hmac.new(self._secret, message, hashlib.sha256).hexdigest()

verify async

verify(memory: Any, *, pinned: bool = False) -> int

Refuse unless memory holds the anchor and its lineage; the anchored seq.

pinned: memory is the eraser's own login, read in its schema.

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

async def verify(self, memory: Any, *, pinned: bool = False) -> int:
    """Refuse unless ``memory`` holds the anchor and its lineage; the anchored seq.

    ``pinned``: ``memory`` is the eraser's own login, read in its schema.
    """
    from symfonic.memory.backends.postgres_erase_binding import MISBOUND
    from symfonic.memory.backends.postgres_eraser_roles import EraserNotReady

    async with _resolving(memory, self._key[1] if pinned else None):
        if await memory.fetchval(_GRAPH) != self._key[1]:  # the eraser erases another graph
            raise EraserNotReady(MISBOUND)

        async def work(conn: Any) -> int:
            return int((await self._lineage(memory, conn, lock=False))["seq"])

        return await self._in_anchor(work)

PostgresEraseAnchor

PostgresEraseAnchor(pool: Any, *, memory: str | None = None)

The anchor row store on its own pool, in a cluster other than memory's.

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

def __init__(self, pool: Any, *, memory: str | None = None) -> None:
    if type(memory) is not str or not memory:
        raise EraseAnchorError(UNNAMED)
    self._pool = pool
    self.memory = memory