Skip to content

symfonic.memory.backends.postgres_accepted_inbox

postgres_accepted_inbox

Bounded ordered discovery of durable PostgreSQL accepted completions.

accepted_batch async 

accepted_batch(journal: PostgresOperationJournal, scope: MemoryScope, after: int, *, limit: int = 500) -> tuple[tuple[CompletedTurn, ...], bool]

Return consecutive authenticated completions and an explicit backlog flag.

Source code in src/symfonic/memory/backends/postgres_accepted_inbox.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
async def accepted_batch(
    journal: PostgresOperationJournal, scope: MemoryScope, after: int, *, limit: int = 500,
) -> tuple[tuple[CompletedTurn, ...], bool]:
    """Return consecutive authenticated completions and an explicit backlog flag."""
    if (type(journal) is not PostgresOperationJournal or type(scope) is not MemoryScope
            or scope.identity_version != 1 or type(after) is not int or after < 0
            or type(limit) is not int or not 1 <= limit <= 500):
        seal.refuse()
    scope.validate()
    levels = json.dumps([list(level) for level in scope.levels], separators=(",", ":"))

    async def snapshot(conn):
        highwater = await conn.fetchval(
            f"SELECT value FROM {completion.SEQUENCES} "
            "WHERE tenant_id=$1 AND scope_levels=$2", scope.tenant, levels,
        )
        rows = await conn.fetch(
            f"SELECT c.invocation_id, c.sequence, j.request_digest "
            f"FROM {completion.COMPLETIONS} c JOIN {codec.JOURNAL} j "
            "ON j.tenant_id=c.tenant_id AND j.invocation_id=c.invocation_id "
            "WHERE c.tenant_id=$1 AND c.scope_levels=$2 AND c.sequence>$3 "
            "ORDER BY c.sequence LIMIT $4", scope.tenant, levels, after, limit + 1,
        )
        return highwater, rows

    try:
        highwater, rows = await journal._unit(snapshot, snapshot=True)
        highwater = 0 if highwater is None else highwater
        if type(highwater) is not int or highwater < after:
            seal.refuse()
        batch = []
        for row in rows[:limit]:
            if row["sequence"] != after + len(batch) + 1:
                seal.refuse()
            durable = await journal.read_completion(
                row["invocation_id"], scope, row["request_digest"],
            )
            if (type(durable) is not CompletedTurn
                    or durable.acceptance_sequence != row["sequence"]):
                seal.refuse()
            batch.append(durable)
        more = len(rows) > limit
        if (not more and after + len(batch) != highwater) or (
                more and after + len(batch) >= highwater):
            seal.refuse()
        return tuple(batch), more
    except MemoryContractError:
        raise
    except Exception as exc:
        raise MemoryContractError("accepted PostgreSQL inbox UNKNOWN") from exc