Skip to content

symfonic.memory.backends.postgres_conversation_restore

postgres_conversation_restore

The validated logical restore of an anchored standalone conversation database (CORE-171).

A restored database is not the anchor's holder, so every guarded checkout refuses MOVED until this procedure moves the anchor to it, once:

  1. On the live source, seal_and_export(conn, proof=log.proof). Every forget must already be acknowledged. It appends a SEAL row and anchors it; from then on the source refuses SEALED. Dump after this.
  2. On the restored database, validate_restore(conn, export, proof=...). Its log must end at the export's sealed row with the anchored digest; a dump from before an acknowledged forget lacks it and refuses BEHIND. The anchor moves here, an OPEN row is appended and anchored. The sealed source, a second restore of the same export and an export the anchor has moved past all refuse.

Both take a raw owner connection, not a guarded pool.

ConversationExport dataclass 

ConversationExport(source: str, seq: int)

The sealed source's holder and the anchored seq of its SEAL row.

seal_and_export async 

seal_and_export(conn: Any, *, proof: ConversationEraseProof) -> ConversationExport

Seal the live source at its anchored head; resumable after a failed acknowledgement.

Source code in src/symfonic/memory/backends/postgres_conversation_restore.py 
57
58
59
60
61
62
63
64
65
66
67
async def seal_and_export(conn: Any, *, proof: ConversationEraseProof) -> ConversationExport:
    """Seal the live source at its anchored head; resumable after a failed acknowledgement."""
    await erase_log.attest(conn)
    row = await proof.in_anchor(lambda anchor: proof.lineage(conn, anchor, lock=False))
    seq, _ = await erase_log.head(conn)
    if not await erase_log.sealed(conn):
        if int(row["seq"]) != seq:
            raise ConversationAnchorError(UNACKED)
        async with conn.transaction():
            await erase_log.log_forget(conn, "", erase_log.SEAL)
    return ConversationExport(str(await conn.fetchval(HOLDER)), await proof.acknowledge_on(conn))

validate_restore async 

validate_restore(conn: Any, export: ConversationExport, *, proof: ConversationEraseProof) -> int

Move the anchor from the sealed source to this restored database, once.

Source code in src/symfonic/memory/backends/postgres_conversation_restore.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def validate_restore(conn: Any, export: ConversationExport, *,
                           proof: ConversationEraseProof) -> int:
    """Move the anchor from the sealed source to this restored database, once."""
    await erase_log.attest(conn)
    if await erase_log.digest_at(conn, export.seq) is None:  # a lineage behind the export
        raise ConversationAnchorError(BEHIND)
    head, _ = await erase_log.head(conn)
    sealed = await erase_log.sealed(conn)
    if not ((sealed and head == export.seq) or (not sealed and head == export.seq + 1)):
        raise ConversationAnchorError(STALE_EXPORT)

    async def work(anchor: Any) -> None:
        holder, schema = await proof.identity(conn, anchor)
        row = await anchor.fetchrow(ROW + " FOR UPDATE", proof.name, schema)
        if row is None:
            raise ConversationAnchorError(UNANCHORED)
        if row["holder"] not in (export.source, holder) or int(row["seq"]) != export.seq:
            raise ConversationAnchorError(STALE_EXPORT)
        chained = await erase_log.digest_at(conn, export.seq)
        if chained is None or not hmac.compare_digest(proof.seal(export.seq, chained),
                                                      row["digest"]):
            raise ConversationAnchorError(BEHIND)
        if holder != export.source:
            await anchor.execute(
                f"UPDATE {ANCHOR_TABLE} SET holder=$3 WHERE memory_system=$1 "
                "AND memory_schema=$2 AND holder=$4 AND seq=$5",
                proof.name, schema, holder, export.source, export.seq)

    await proof.in_anchor(work)
    if sealed:
        async with conn.transaction():
            await erase_log.log_forget(conn, "", erase_log.OPEN)
    return await proof.acknowledge_on(conn)