Skip to content

symfonic.memory.backends.postgres_accepted_handoff

postgres_accepted_handoff

Factory-bound native completion handoff with opt-in durable QUICK.

PostgresAcceptedHandoff 

PostgresAcceptedHandoff(journal: PostgresOperationJournal, domain: PostgresReceiptDomain, scope: MemoryScope)

Authenticated acceptance consumer bound to one issued receipt composition.

Source code in src/symfonic/memory/backends/postgres_accepted_handoff.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(
    self, journal: PostgresOperationJournal, domain: PostgresReceiptDomain, scope: MemoryScope
):
    if (
        type(journal) is not PostgresOperationJournal
        or type(domain) is not PostgresReceiptDomain
        or type(scope) is not MemoryScope
        or scope.identity_version != 1
    ):
        seal.refuse()
    journal.require_domain(domain)
    scope.validate()
    self._journal, self._domain, self._scope = journal, domain, scope
    self._cadence = None

accept async 

accept(completion: CompletedTurn) -> CompletedTurn

Authenticate the persisted decision; caller values grant no authority.

Source code in src/symfonic/memory/backends/postgres_accepted_handoff.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
async def accept(self, completion: CompletedTurn) -> CompletedTurn:
    """Authenticate the persisted decision; caller values grant no authority."""
    if type(completion) is not CompletedTurn or completion.full_scope != self._scope:
        seal.refuse()
    durable = await self._journal.read_completion(
        completion.invocation_id,
        self._scope,
        completion.request_digest,
    )
    if type(durable) is not CompletedTurn or durable != completion:
        seal.refuse()
    if self._cadence is not None:
        _, more = await self.recover_quick()
        if more:
            raise MemoryContractError(
                "accepted cadence UNKNOWN; recover bounded backlog without body replay")
    return durable

bind_quick async 

bind_quick(coordinator, *, policy_digest: str)

Opt in this issued callback to the exact durable QUICK policy.

Source code in src/symfonic/memory/backends/postgres_accepted_handoff.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
async def bind_quick(self, coordinator, *, policy_digest: str):
    """Opt in this issued callback to the exact durable QUICK policy."""
    if self._cadence is not None:
        seal.refuse()
    from symfonic.memory.backends.postgres_native_accepted_cadence import (
        PostgresNativeAcceptedCadence,
    )

    cadence = PostgresNativeAcceptedCadence(
        self,
        coordinator,
        policy_digest=policy_digest,
    )
    self._cadence = cadence
    try:
        await cadence.ensure_schema()
    except BaseException:
        self._cadence = None
        raise
    return cadence

recover async 

recover(after: int, *, limit: int = 500)

Discover ordered acceptance after callback loss without body replay.

Source code in src/symfonic/memory/backends/postgres_accepted_handoff.py 
79
80
81
async def recover(self, after: int, *, limit: int = 500):
    """Discover ordered acceptance after callback loss without body replay."""
    return await accepted_batch(self._journal, self._scope, after, limit=limit)

recover_quick async 

recover_quick(*, limit: int = 500) -> tuple[int, bool]

Consume at most one validated inbox batch, never a turn body.

Source code in src/symfonic/memory/backends/postgres_accepted_handoff.py 
73
74
75
76
77
async def recover_quick(self, *, limit: int = 500) -> tuple[int, bool]:
    """Consume at most one validated inbox batch, never a turn body."""
    if self._cadence is None or type(limit) is not int or not 1 <= limit <= 500:
        seal.refuse()
    return await self._cadence.recover_scope(self._scope, limit=limit)