Skip to content

symfonic.memory.backends.postgres_journal_completion

postgres_journal_completion

Atomic v1 accepted completion for the PostgreSQL invocation journal.

One journal unit locks the exact owner/revision claim, authenticates every prepared operation envelope, requires each FLUSH publication's exact COMMITTED receipt in the same receipt domain, allocates the per-scope acceptance sequence, seals both into the completion and advances the revision. Any refusal rolls the whole unit back. A publication that is not provably COMMITTED, or an ambiguous commit, is UNKNOWN: the instance refuses further work and nothing is retried. v0 identities are refused by the codec.

prepared_operations async 

prepared_operations(conn, cipher, expected, scope)

Authenticate and restore every prepared operation of one claim.

Source code in src/symfonic/memory/backends/postgres_journal_completion.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
async def prepared_operations(conn, cipher, expected, scope):
    """Authenticate and restore every prepared operation of one claim."""
    rows = await conn.fetch(
        f"SELECT phase, sealed FROM {codec.OPERATIONS} "
        "WHERE tenant_id=$1 AND invocation_id=$2",
        expected["tenant"], expected["invocation"],
    )
    operations = []
    for row in rows:
        envelope = json.loads(cipher.decrypt(bytes(row["sealed"])))
        if (type(envelope) is not list or len(envelope) != 2
                or type(envelope[0]) is not list or len(envelope[0]) != 2
                or envelope[0][0] != "operation"):
            seal.refuse()
        binding = envelope[0][1]
        if (type(binding) is not list or len(binding) != 3 or binding[0] != expected
                or seal.phase(binding[1], binding[2]) != row["phase"]):
            seal.refuse()
        operations.append(seal.restore(cipher, binding, row["sealed"], scope))
    return operations

verify_owner async 

verify_owner(journal, owner, expected_revision)

Snapshot exact owner and revision; a completed claim stays verifiable.

Source code in src/symfonic/memory/backends/postgres_journal_completion.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def verify_owner(journal, owner, expected_revision):
    """Snapshot exact owner and revision; a completed claim stays verifiable."""
    from symfonic.capabilities.memory.owner_verifier import OwnerVerified

    expected, verifier, revision = journal._owner_binding(owner, expected_revision)

    async def work(conn):
        row = await codec.fetch(conn, expected)
        return row is not None and secrets.compare_digest(row["owner"], verifier) and (
            row["revision"] == revision)

    try:
        owned = await journal._unit(work, snapshot=True)
    except MemoryContractError:
        raise
    except Exception:
        journal._uncertain = True
        return JournalUnknown()
    if not owned:
        seal.refuse()
    return OwnerVerified()