PostgresOperationJournal(domain, *, cipher, staged_publication=False, rendered_selection_acknowledgement=False, linked_selection_acknowledgement=False, linked_edges=None)
Owner-CAS journal with gated v2/v3 selection and staged v1.
Staged publication excludes v2/v3 writers; old readers refuse new envelopes.
Staged v1 publication excludes the separately gated v2/v3 writers.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| def __init__(self, domain, *, cipher, staged_publication=False,
rendered_selection_acknowledgement=False,
linked_selection_acknowledgement=False, linked_edges=None) -> None:
"""Staged v1 publication excludes the separately gated v2/v3 writers."""
from symfonic.memory.backends.postgres_linked_edges_v1 import PostgresLinkedEdgesV1
from symfonic.memory.backends.postgres_operation_receipt import PostgresReceiptDomain
if (type(domain) is not PostgresReceiptDomain
or type(staged_publication) is not bool
or type(rendered_selection_acknowledgement) is not bool
or type(linked_selection_acknowledgement) is not bool
or (staged_publication and (rendered_selection_acknowledgement
or linked_selection_acknowledgement))
or (linked_selection_acknowledgement and (
not rendered_selection_acknowledgement
or type(linked_edges) is not PostgresLinkedEdgesV1
or linked_edges.pool is not domain.pool))
or (not linked_selection_acknowledgement and linked_edges is not None)):
seal.refuse()
self._domain = domain
self._cipher = seal.require_cipher(cipher)
self._uncertain = False
self.staged_publication = staged_publication
self._rendered_selection_acknowledgement = rendered_selection_acknowledgement
self._linked_selection_acknowledgement = linked_selection_acknowledgement
self._linked_edges = linked_edges
|
complete_invocation
async
complete_invocation(owner, expected_revision, authority, publications)
Atomically accept the turn: committed FLUSH references plus sequence.
A staged journal never accepts: its turns end READY, and only a
composed PostgresStagedPublisher publishes them.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def complete_invocation(self, owner, expected_revision, authority, publications):
"""Atomically accept the turn: committed FLUSH references plus sequence.
A staged journal never accepts: its turns end READY, and only a
composed ``PostgresStagedPublisher`` publishes them.
"""
if self.staged_publication:
seal.refuse()
return await completion.complete(
self, owner, expected_revision, authority, publications,
)
|
ensure_schema
async
Default journals install no staged table and no claim fence.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def ensure_schema(self) -> None:
"""Default journals install no staged table and no claim fence."""
statements = (*codec.DDL, *completion.DDL)
if self.staged_publication:
from symfonic.memory.backends import postgres_journal_staged as staged
statements = (*statements, *staged.DDL)
async def create(conn):
for statement in statements:
await conn.execute(statement)
await self._unit(create)
|
has_invocation
async
has_invocation(invocation_id, full_scope, request_digest)
Read-only exact-identity check; a different request refuses.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def has_invocation(self, invocation_id, full_scope, request_digest):
"""Read-only exact-identity check; a different request refuses."""
expected = codec.identity(invocation_id, full_scope, request_digest)
async def work(conn):
return await codec.fetch(conn, expected) is not None
return await self._unit(work, snapshot=True)
|
outcome
async
Only an exact COMMITTED receipt in this same domain is positive.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def outcome(self, operation):
"""Only an exact COMMITTED receipt in this same domain is positive."""
if (
self._uncertain
or type(operation) is not MemoryOperation
or operation.scope.identity_version != 1
):
return MemoryOperationOutcome.UNKNOWN
return await self._domain.reconcile(operation)
|
ready
async
ready(owner, expected_revision, authority, operation, output_sha256)
Seal READY: committed WRITEs plus one prepared, never-run FLUSH.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def ready(self, owner, expected_revision, authority, operation, output_sha256):
"""Seal READY: committed WRITEs plus one prepared, never-run FLUSH."""
from symfonic.memory.backends import postgres_journal_staged as staged
staged.require_staged(self)
return await staged.ready(
self, owner, expected_revision, authority, operation, output_sha256,
)
|
require_domain
require_domain(domain) -> None
Refuse any receipt domain other than the one this journal shares.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| def require_domain(self, domain) -> None:
"""Refuse any receipt domain other than the one this journal shares."""
if domain is not self._domain or domain.pool is not self._domain.pool:
seal.refuse()
|
snapshot
async
snapshot(invocation_id, full_scope)
Read-only STAGED/READY/ACTIVE/UNKNOWN position of one staged turn.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def snapshot(self, invocation_id, full_scope):
"""Read-only STAGED/READY/ACTIVE/UNKNOWN position of one staged turn."""
from symfonic.memory.backends import postgres_journal_staged as staged
staged.require_staged(self)
return await staged.snapshot(self, invocation_id, full_scope)
|
verify_owner
async
verify_owner(owner, expected_revision)
Snapshot exact owner/revision; no write, no revision advance.
Source code in src/symfonic/memory/backends/postgres_operation_journal.py
| async def verify_owner(self, owner, expected_revision):
"""Snapshot exact owner/revision; no write, no revision advance."""
return await completion.verify_owner(self, owner, expected_revision)
|