Skip to content

symfonic.memory.backends.postgres_journal_seal

postgres_journal_seal

Authenticated sealing and validation for the PostgreSQL invocation journal.

The cipher is an externally owned authenticated-encryption collaborator with encrypt(bytes)->bytes and decrypt(bytes)->bytes (for example Fernet). Every sealed value carries its claim and phase binding, so a ciphertext copied to another row or phase refuses instead of decoding.

checkpoint 

checkpoint(cipher, binding, row, phase_key, ordinal)

Decode one COMPLETE checkpoint row; any other state refuses.

Source code in src/symfonic/memory/backends/postgres_journal_seal.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def checkpoint(cipher, binding, row, phase_key, ordinal):
    """Decode one COMPLETE checkpoint row; any other state refuses."""
    if row["state"] != "COMPLETE" or row["sealed"] is None:
        refuse()
    payload = unseal(cipher, binding, row["sealed"])
    if type(payload) is not dict or set(payload) != {"output", "ids", "revision", "manifest"}:
        refuse()
    from symfonic.capabilities.memory.selected_records import SelectedRecordManifest

    manifest = payload["manifest"]
    manifest = SelectedRecordManifest.from_document(manifest) if manifest is not None else None
    if type(payload["ids"]) is not list:
        refuse()
    result = CompleteCheckpoint(
        phase_key,
        ordinal,
        base64.b64decode(payload["output"], validate=True),
        tuple(payload["ids"]),
        revision(payload["revision"]),
        selected_manifest=manifest,
    )
    if result.revision < 1:
        refuse()
    return result

seal_operation 

seal_operation(cipher, binding, operation, body)

Encrypt the canonical body too, so restart restores the exact selection.

Source code in src/symfonic/memory/backends/postgres_journal_seal.py 
121
122
123
124
125
126
127
128
129
def seal_operation(cipher, binding, operation, body):
    """Encrypt the canonical body too, so restart restores the exact selection."""
    return seal(cipher, ["operation", binding], {
        "id": operation.operation_id,
        "verb": operation.verb.value,
        "content": operation.content_digest,
        "intent": operation.intent_digest,
        "body": body,
    })

settle async 

settle(journal, awaitable)

Finish in-flight database I/O before cancellation propagates.

A cancelled unit may still have committed, so the journal instance is marked uncertain and refuses further work; a fresh instance must re-read.

Source code in src/symfonic/memory/backends/postgres_journal_seal.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
async def settle(journal, awaitable):
    """Finish in-flight database I/O before cancellation propagates.

    A cancelled unit may still have committed, so the journal instance is
    marked uncertain and refuses further work; a fresh instance must re-read.
    """
    future = asyncio.ensure_future(awaitable)
    try:
        return await asyncio.shield(future)
    except asyncio.CancelledError:
        journal._uncertain = True
        while not future.done():
            try:
                await asyncio.shield(future)
            except asyncio.CancelledError:
                continue
            except Exception:
                break
        if not future.cancelled():
            future.exception()  # Retrieved: the caller sees only cancellation.
        raise