symfonic.memory.backends.postgres_memory_restore

postgres_memory_restore

The restore procedure for the memory erase generation table (CORE-171).

A restored database holds content and generations as they were at its backup: any forget since is undone in both, so neither a stale turn's token nor a fresh turn's capture can see that it happened. Every logical restore -- a dump restored into this cluster or another, or pg_upgrade -- leaves a witness that no longer binds (postgres_memory_generation_witness), so every generation-table reader -- Agent turn admission and the erase gates -- refuses, and the install script refuses to re-mint it. A direct backend read outside a turn (get_node, query_subtree) does not consult the witness yet: its coverage is pending, not excluded. Only this procedure re-opens such a table:

1. :func:seal_and_export on the source, as memory_nodes' owner: its witness must still bind, so the source is the live table and not itself a restore. Under an EXCLUSIVE lock it reads every generation row, then seals the table: no reader binds it again and the forward function refuses every write, including a forget already waiting on the lock. The export names every erasure the source ever recorded, and nothing can add one.
2. Restore the database, content and generations from one consistent snapshot, then :func:validate_restore on it, as its memory_nodes' owner. Its generation rows must equal the export's exactly. Fewer or lower rows mean the backup predates a forget, whose content it restored: that refuses, and the table stays closed. Its row trigger must also carry this export's seal, or its source's witness (a snapshot from before the seal): an export of another lineage refuses, however self-consistent. The digest detects an edited export; it authenticates nothing, and nothing here proves the export is the source's latest -- that freshness needs proof kept outside the database. Equal rows then mean every forget the export recorded is in the restored data; it mints a new witness.

Since 11.3.0 both steps also take the eraser's erase anchor proof (postgres_erase_anchor), which supplies the freshness and single use the export cannot. The seal refuses unless the anchor holds this source at its log head (every forget acknowledged); the export records that holder and seq. The validation refuses unless the restored log carries row seq under the anchored digest, and, in the transaction that mints, rebinds the anchor from the source to the restored database: the sealed source, a second restore of the same export and an export the anchor has moved past all refuse. A physical copy keeps the holder; the anchor's lineage check refuses one that lacks an acknowledged forget, at every connection checkout.

GenerationExport dataclass

GenerationExport(source_system: str, source_witness: int, rows: tuple[tuple[str, str, int], ...], digest: str, holder: str = '', log_seq: int = -1)

Every generation row of a sealed source, and the witness it sealed.

seal_and_export async

seal_and_export(conn: Any, *, proof: Any = None) -> GenerationExport

Seal the live source table and return every generation it recorded.

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

async def seal_and_export(conn: Any, *, proof: Any = None) -> GenerationExport:
    """Seal the live source table and return every generation it recorded."""
    from symfonic.memory.backends.postgres_memory_erase_log import head

    if proof is None:
        raise MemoryContractError(UNPROVEN)
    async with conn.transaction():
        await _owner(conn)
        table = await require_table(conn)  # UNWITNESSED: a restore cannot be a source
        witness = int(await conn.fetchval(f"SELECT {witness_sql('$1::regclass')}", table))
        rows = await _rows(conn, table)
        system = await conn.fetchval(_SYSTEM)
        log_seq = (await head(conn))[0]  # the log lock waits out any forget in flight
        if await proof.verify(conn) != log_seq:
            raise MemoryContractError(UNACKNOWLEDGED)
        holder = await proof.holder(conn)
        digest = _digest(system, witness, rows, holder, log_seq)
        await conn.execute(
            f"DROP TRIGGER {FORWARD}_rows ON {table}; "
            f"CREATE TRIGGER {FORWARD}_rows BEFORE INSERT OR UPDATE OR DELETE ON {table} "
            f"FOR EACH ROW EXECUTE FUNCTION {FORWARD}('{SEALED}', '{digest}'); "
            f"ALTER TABLE {table} ENABLE ALWAYS TRIGGER {FORWARD}_rows"
        )
        return GenerationExport(system, witness, rows, digest, holder, log_seq)

validate_restore async

validate_restore(conn: Any, export: GenerationExport, *, proof: Any = None) -> int

Re-open a restored table whose rows equal export's; the new witness.

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

async def validate_restore(conn: Any, export: GenerationExport, *, proof: Any = None) -> int:
    """Re-open a restored table whose rows equal ``export``'s; the new witness."""
    if proof is None or export.log_seq < 0:
        raise MemoryContractError(UNPROVEN)
    if _digest(export.source_system, export.source_witness, export.rows, export.holder,
               export.log_seq) != export.digest:
        raise MemoryContractError(TAMPERED)
    async with conn.transaction():
        await _owner(conn)
        table, _, _, guarded, *_ = await conn.fetchval(RESOLVE, FORWARD_BODY, list(CONSTRAINTS))
        if table is None or not guarded:
            raise MemoryContractError(UNGUARDED)
        raw = await conn.fetchval(_ARGS, table)
        arguments = [arg.decode() for arg in (raw or b"").split(b"\x00")[:-1]]
        if arguments not in (  # sealed by it, or a snapshot of its source before the seal
            [SEALED, export.digest], [str(export.source_witness), export.source_system]
        ):
            raise MemoryContractError(LINEAGE)
        rows = dict(((t, p), g) for t, p, g in await _rows(conn, table))
        source = dict(((t, p), g) for t, p, g in export.rows)
        if any(rows.get(key, 0) > generation for key, generation in source.items()) or (
            rows.keys() - source.keys()
        ):
            raise MemoryContractError(FOREIGN)
        behind = sum(1 for key, generation in source.items() if rows.get(key) != generation)
        if behind:
            raise MemoryContractError(BEHIND.format(count=behind))
        log = table.rsplit(".", 1)[0] + ".memory_erase_log"  # read before the mint
        if await conn.fetchval(f"SELECT max(seq) FROM {log}") != export.log_seq:
            raise MemoryContractError(LINEAGE)  # the export is the log's whole history
        witness = await mint(conn)
        await proof.rebind(conn, source=export.holder, seq=export.log_seq)  # last, then commit
        return witness