symfonic.memory.backends.postgres_memory_erase_log

postgres_memory_erase_log

The serialized, hash-chained erase log beside the generations (CORE-171).

A physical copy of the memory cluster cannot be told from the live cluster from inside it, so the proof that a database holds every acknowledged forget is the log's head, anchored outside the backup domain (postgres_erase_anchor). This module is the in-database half.

Every generation advance appends one row in the erase's own transaction. The append trigger locks the log, so log order is commit order, assigns seq = head + 1 and digest = sha256(prior digest || row), where the row includes its transaction ID and a nonce the trigger draws (callers cannot choose either): row N's digest commits to the full prefix history 0..N, so two lineages that differ anywhere below N cannot share it, and a copy replaying the very same forgets, through any login, cannot reproduce it. Row 0 is written by the install and binds the generation rows present then.

The append function runs as the log's owner (SECURITY DEFINER, search_path pinned to pg_catalog, EXECUTE revoked from PUBLIC): an INSERT-only login may append, only the owner may take the serializing lock, and no caller's search_path reaches its body. Every check attests both triggers' exact function (owner, body, pinned search_path) before it trusts the log.

A deferred constraint trigger on the generations refuses, at commit, any generation insert or update without its log row from the same transaction: an earlier release's advance SQL fails closed and deletes nothing. The log admits no update, delete or truncate. A database owner or superuser can still rewrite both; that is the stated tamper boundary.

advance async

advance(conn: Any, scope: Any) -> int

Advance scope's generation and append its log row; return its seq.

Inside the erase's own transaction, after the graph lock and before it commits. The append waits on the log lock, so concurrent forgets of one memory database commit one at a time.

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

async def advance(conn: Any, scope: Any) -> int:
    """Advance ``scope``'s generation and append its log row; return its seq.

    Inside the erase's own transaction, after the graph lock and before it
    commits. The append waits on the log lock, so concurrent forgets of one
    memory database commit one at a time.
    """
    from symfonic.memory.backends.postgres_erase_jobs import canonical_job_prefix
    from symfonic.memory.backends.postgres_memory_generation import require_table

    table = await require_table(conn)
    log = await log_table(conn)
    prefix = scope.subtree_prefix
    if not canonical_job_prefix(scope.tenant_id, prefix):
        raise MemoryContractError("memory erase generation prefix is not canonical")
    generation = await conn.fetchval(
        f"INSERT INTO {table} (tenant_id, prefix, generation) VALUES ($1, $2, 1) "
        f"ON CONFLICT (tenant_id, prefix) DO UPDATE SET generation = {table}.generation + 1 "
        "RETURNING generation",
        scope.tenant_id,
        prefix,
    )
    return await conn.fetchval(
        f"INSERT INTO {log} (seq, tenant_id, prefix, generation, xid, nonce, digest) "
        "VALUES (0, $1, $2, $3, '0'::xid8, gen_random_uuid(), ''::bytea) RETURNING seq",
        scope.tenant_id,
        prefix,
        generation,
    )

digest_at async

digest_at(conn: Any, seq: int) -> bytes | None

Row seq's chained digest, or None when this database lacks it.

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

async def digest_at(conn: Any, seq: int) -> bytes | None:
    """Row ``seq``'s chained digest, or None when this database lacks it."""
    value = await conn.fetchval(f"SELECT digest FROM {await log_table(conn)} WHERE seq=$1", seq)
    return None if value is None else bytes(value)

head async

head(conn: Any) -> tuple[int, bytes]

The committed head (seq, digest) this connection sees.

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

async def head(conn: Any) -> tuple[int, bytes]:
    """The committed head ``(seq, digest)`` this connection sees."""
    row = await conn.fetchrow(
        f"SELECT seq, digest FROM {await log_table(conn)} ORDER BY seq DESC LIMIT 1"
    )
    if row is None:
        raise MemoryContractError(UNLOGGED)
    return int(row["seq"]), bytes(row["digest"])

log_table async

log_table(conn: Any) -> str

The log beside memory_erase_generations, or refuse UNLOGGED.

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

async def log_table(conn: Any) -> str:
    """The log beside ``memory_erase_generations``, or refuse ``UNLOGGED``."""
    from symfonic.memory.backends.postgres_memory_generation import require_table

    schema = (await require_table(conn)).rsplit(".", 1)[0]
    if not await conn.fetchval(
        _INSTALLED, f"{schema}.memory_erase_generations", f"{schema}.{_LOGGED}()",
        _LOGGED_BODY, f"{schema}.{LOG_TABLE}", f"{schema}.{_APPEND}()", _APPEND_BODY,
    ):
        raise MemoryContractError(UNLOGGED)
    return f"{schema}.{LOG_TABLE}"