symfonic.memory.backends.postgres_memory_generation

postgres_memory_generation

Durable memory erase generations, in the memory database itself (CORE-199).

A turn that recalled memory before an admin forget holds erased content and may write it back after the forget returned. Every native subtree erase -- the runtime's inline forget, or the eraser's first transaction of a large job -- therefore advances a row for the erased key in the transaction that deletes the content, under the tenant lock and the sweep's EXCLUSIVE lock. A memory-only forget advances it as a joint forget does.

Every raw PostgreSQL v1 mutation re-reads the rows covering its own scope on its own connection after the tenant lock: the scope's key and each ancestor key, by primary key. A turn's capture, and every re-proof of it, also reads every key below its scope by a primary-key range: a turn recalls its whole subtree, so a forget of a descendant erased something it may hold. That read is bounded: past SUBTREE_BUDGET keys it refuses rather than prove less. The rows live beside memory_nodes, so the proof needs no process-local registry and holds across pools, processes and restarts. Rows only advance: a trigger admits an INSERT only at generation 1 and an UPDATE only to exactly the old generation plus one, and refuses every DELETE or TRUNCATE, so a role holding INSERT and UPDATE (the runtime and the eraser) can neither reopen a stale turn nor exhaust a counter. The triggers are ENABLE ALWAYS, so session_replication_role=replica still fires them. A turn's token also binds the table's install witness and the server (INCARNATION, SERVER; postgres_memory_generation_witness): a reinstalled, restored or copied table, or a restored cluster, must not readmit a turn captured before, even if its rows match. Every read refuses a table whose triggers are not enabled always, carry another event mask, a WHEN clause or an UPDATE OF column list, or call any function but this module's body beside the table, a table with row security, a rewrite rule, any other trigger or an inheritance child or parent, an UNLOGGED table, and a table or trigger function that memory_nodes' owner does not own (postgres_memory_generation_schema). The table owner, or a superuser, can still tamper: a persistent change of those kinds makes every reader refuse, but a trigger disabled and re-enabled inside one transaction passes that transaction's own writes (afterwards its row no longer matches the witness, UNWITNESSED), and nothing here can prevent it.

Startup never creates the table; scripts/migrations/core199_memory_erase_generations.sql does. Until then a turn refuses to capture and a native erase refuses with UNMIGRATED before it deletes anything (CORE-171): the eraser does not attest, the runtime's forget refuses before its first delete and a joint forget before its transcript intent (a check at its start, not a lock). An erase that could not record would leave a turn free to write back what it erased.

advance async

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

Invalidate every turn that observed scope's subtree; return its log seq.

Inside the erase's own transaction, after the graph lock and before it commits, whether or not the erase matched a row. Callers refuse earlier; a table or log gone since then refuses here and rolls the delete back. The generation and its erase log row commit together (postgres_memory_erase_log).

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

async def advance(conn: Any, scope: Any) -> int:
    """Invalidate every turn that observed ``scope``'s subtree; return its log seq.

    Inside the erase's own transaction, after the graph lock and before it
    commits, whether or not the erase matched a row. Callers refuse earlier;
    a table or log gone since then refuses here and rolls the delete back.
    The generation and its erase log row commit together
    (``postgres_memory_erase_log``).
    """
    from symfonic.memory.backends.postgres_memory_erase_log import advance as logged

    return await logged(conn, scope)

capture async

capture(pool: Any, path: str, *, strict: bool) -> tuple[tuple[str, int], ...]

The rows covering path now; a turn takes this before it reads.

ROW SHARE on memory_nodes waits out an erase holding the sweep's EXCLUSIVE lock, so a table installed during that erase cannot hide it. strict (every turn) refuses without the table; otherwise it reads ().

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

async def capture(pool: Any, path: str, *, strict: bool) -> tuple[tuple[str, int], ...]:
    """The rows covering ``path`` now; a turn takes this before it reads.

    ROW SHARE on ``memory_nodes`` waits out an erase holding the sweep's
    EXCLUSIVE lock, so a table installed during that erase cannot hide it.
    ``strict`` (every turn) refuses without the table; otherwise it reads ().
    """
    try:
        async with pool.transaction():
            conn = await pool.acquire()
            try:
                table = await generations_table(conn)
                if table is None:
                    if strict:
                        raise MemoryContractError(UNMIGRATED)
                    return ()
                schema = table.rsplit(".", 1)[0]
                await conn.execute(f"LOCK TABLE {schema}.memory_nodes IN ROW SHARE MODE")
                return await covering(conn, path, subtree=True) or ()
            finally:
                await pool.release(conn)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("memory erase generation could not be read") from exc

checked async

checked(pool: Any, path: str, *, subtree: bool = False) -> tuple[tuple[str, int], ...] | None

:func:covering in the pool's (joined) transaction, failing closed.

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

async def checked(
    pool: Any,
    path: str,
    *,
    subtree: bool = False,
) -> tuple[tuple[str, int], ...] | None:
    """:func:`covering` in the pool's (joined) transaction, failing closed."""
    try:
        async with pool.transaction():
            conn = await pool.acquire()
            try:
                return await covering(conn, path, subtree=subtree)
            finally:
                await pool.release(conn)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - fail closed, hide backend details
        raise MemoryUnavailable("memory erase generation could not be checked") from exc

covering async

covering(conn: Any, path: str, *, subtree: bool = False) -> tuple[tuple[str, int], ...] | None

Rows at path's key and each ancestor key, by primary key; None if absent.

subtree adds every key below path's: what a turn that recalled path read, at most SUBTREE_BUDGET of them, or it refuses with OVER_BUDGET. That form is a turn's token, so it leads with the table's INCARNATION and the SERVER: neither a reinstalled or restored table nor a restored cluster readmits a turn captured before. Every row must come from the relation RESOLVE attested, and the name must still resolve to it after the reads, or it refuses REBOUND. The caller's transaction holds the tenant lock; a joined caller must be READ COMMITTED, or its snapshot could predate an advance it waited for.

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

async def covering(
    conn: Any,
    path: str,
    *,
    subtree: bool = False,
) -> tuple[tuple[str, int], ...] | None:
    """Rows at ``path``'s key and each ancestor key, by primary key; None if absent.

    ``subtree`` adds every key below ``path``'s: what a turn that recalled
    ``path`` read, at most ``SUBTREE_BUDGET`` of them, or it refuses with
    ``OVER_BUDGET``. That form is a turn's token, so it leads with the
    table's ``INCARNATION`` and the ``SERVER``: neither a reinstalled or
    restored table nor a restored cluster readmits a turn captured before.
    Every row must come from the relation ``RESOLVE`` attested, and the name
    must still resolve to it after the reads, or it refuses ``REBOUND``.
    The caller's transaction holds the tenant lock; a joined caller must be
    READ COMMITTED, or its snapshot could predate an advance it waited for.
    """
    table, isolation, relation = await _resolve(conn)
    if table is None:
        return None
    if isolation != "read committed":
        raise MemoryContractError(ISOLATION)
    scope = _storage(path)
    from symfonic.memory.backends.postgres_erase_retirement import refuse_admission

    await refuse_admission(conn, scope.tenant_id)
    rows = list(
        await conn.fetch(
            COVERING.format(table=table), scope.tenant_id, scope.ancestor_prefix_paths()
        )
    )
    if subtree:
        prefix = scope.subtree_prefix
        below = await conn.fetch(
            BELOW.format(table=table),
            scope.tenant_id,
            prefix,
            prefix[:-1] + "0",
            SUBTREE_BUDGET + 1,
        )
        if len(below) > SUBTREE_BUDGET:
            raise MemoryContractError(OVER_BUDGET)
        rows += below
    witness, server = await bound_witness(conn, table, relation, rows)
    pairs = tuple(sorted((row["prefix"], row["generation"]) for row in rows))
    return ((INCARNATION, witness), (SERVER, server), *pairs) if subtree else pairs

generations_table async

generations_table(conn: Any) -> str | None

The schema-qualified table beside memory_nodes, or None.

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

async def generations_table(conn: Any) -> str | None:
    """The schema-qualified table beside ``memory_nodes``, or None."""
    return (await _resolve(conn))[0]

grant_runtime async

grant_runtime(conn: Any, table: str, role: str) -> None

Owner side, in the caller's transaction; role is a validated identifier.

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

async def grant_runtime(conn: Any, table: str, role: str) -> None:
    """Owner side, in the caller's transaction; ``role`` is a validated identifier."""
    await conn.execute(f'REVOKE ALL ON {table} FROM "{role}"')
    await conn.execute(f'GRANT {", ".join(RUNTIME_GRANTS)} ON {table} TO "{role}"')
    if drift := await runtime_drift(conn, table, role):
        raise MemoryContractError(drift)
    log = table.rsplit(".", 1)[0] + ".memory_erase_log"  # CORE-171: checkouts read, forgets append
    if await conn.fetchval("SELECT to_regclass($1)", log) is not None:
        await conn.execute(f'REVOKE ALL ON {log} FROM "{role}"')
        await conn.execute(f'GRANT SELECT, INSERT ON {log} TO "{role}"')

require_installed async

require_installed(pool: Any) -> None

:func:require_table on pool before an erase touches another store.

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

async def require_installed(pool: Any) -> None:
    """:func:`require_table` on ``pool`` before an erase touches another store."""
    try:
        conn = await pool.acquire()
        try:
            await require_table(conn)
        finally:
            await pool.release(conn)
    except MemoryContractError:
        raise
    except Exception as exc:  # noqa: BLE001 - one unavailable-store contract
        raise MemoryUnavailable("memory erase generation could not be read") from exc

require_table async

require_table(conn: Any) -> str

The table, or refuse UNMIGRATED: no native erase runs without it.

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

async def require_table(conn: Any) -> str:
    """The table, or refuse ``UNMIGRATED``: no native erase runs without it."""
    table = await generations_table(conn)
    if table is None:
        raise MemoryContractError(UNMIGRATED)
    return table

runtime_drift async

runtime_drift(conn: Any, table: str, role: str) -> str | None

How role departs from read-and-advance on table, or None.

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

async def runtime_drift(conn: Any, table: str, role: str) -> str | None:
    """How ``role`` departs from read-and-advance on ``table``, or None."""
    for privilege in RUNTIME_GRANTS + _RUNTIME_EXCESS:
        wanted = privilege in RUNTIME_GRANTS
        held = await conn.fetchval(HELD, role, table, privilege, not wanted, list(COLUMN_KINDS))
        if held is not wanted:
            return f"{role} {'lacks' if not held else 'holds'} {privilege} on {table}"
    return None