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
| 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
| 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
| 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
|