Skip to content

symfonic.memory.backends.postgres_operation_journal_codec

postgres_operation_journal_codec

Validated PostgreSQL v1 invocation journal identity, schema and row access.

advance async 

advance(conn, expected, verifier, revision)

Owner and expected-revision compare-and-swap inside the caller's unit.

Source code in src/symfonic/memory/backends/postgres_operation_journal_codec.py 
143
144
145
146
147
148
149
150
151
async def advance(conn, expected, verifier, revision):
    """Owner and expected-revision compare-and-swap inside the caller's unit."""
    changed = await conn.execute(
        f"UPDATE {JOURNAL} SET revision=revision+1 WHERE tenant_id=$1 "
        "AND invocation_id=$2 AND owner=$3 AND revision=$4",
        expected["tenant"], expected["invocation"], verifier, revision,
    )
    if changed != "UPDATE 1":
        seal.refuse()

fetch async 

fetch(conn, expected, *, lock=False)

The claim row for this key, refused unless its identity is exact.

Source code in src/symfonic/memory/backends/postgres_operation_journal_codec.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def fetch(conn, expected, *, lock=False):
    """The claim row for this key, refused unless its identity is exact."""
    row = await conn.fetchrow(
        f"SELECT * FROM {JOURNAL} WHERE tenant_id=$1 AND invocation_id=$2"
        + (" FOR UPDATE" if lock else ""),
        expected["tenant"], expected["invocation"],
    )
    if row is None:
        return None
    if (
        row["schema"] != expected["schema"]
        or _levels(row["scope_levels"]) != expected["levels"]
        or row["request_digest"] != expected["request"]
    ):
        seal.refuse()
    seal.revision(row["revision"])
    seal.digest(row["owner"])
    return row

identity 

identity(invocation_id, scope, request_digest)

Exact v1 claim identity; positional v0 scopes carry no native kinds.

Source code in src/symfonic/memory/backends/postgres_operation_journal_codec.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def identity(invocation_id, scope, request_digest):
    """Exact v1 claim identity; positional v0 scopes carry no native kinds."""
    if (
        type(invocation_id) is not str
        or not 1 <= len(invocation_id) <= 128
        or not invocation_id.isascii()
        or not invocation_id.isprintable()
        or type(scope) is not MemoryScope
    ):
        seal.refuse()
    scope.validate()
    if scope.identity_version != 1:
        seal.refuse()
    return {
        "schema": SCHEMA,
        "tenant": scope.tenant,
        "invocation": invocation_id,
        "levels": [list(level) for level in scope.levels],
        "request": seal.digest(request_digest),
    }

insert_claim async 

insert_claim(conn, expected, verifier)

Atomic first insert; True only for the claimant that created it.

Source code in src/symfonic/memory/backends/postgres_operation_journal_codec.py 
115
116
117
118
119
120
121
122
123
124
async def insert_claim(conn, expected, verifier):
    """Atomic first insert; ``True`` only for the claimant that created it."""
    created = await conn.fetchval(
        f"INSERT INTO {JOURNAL} (tenant_id, invocation_id, schema, scope_levels, "
        "request_digest, owner, revision) VALUES ($1,$2,$3,$4::jsonb,$5,$6,0) "
        "ON CONFLICT DO NOTHING RETURNING 1",
        expected["tenant"], expected["invocation"], expected["schema"],
        json.dumps(expected["levels"]), expected["request"], verifier,
    )
    return created == 1

owned async 

owned(conn, expected, verifier, revision)

Lock the claim; refuse unless owner verifier and revision both match.

An accepted completion closes the claim: no later phase may be written.

Source code in src/symfonic/memory/backends/postgres_operation_journal_codec.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def owned(conn, expected, verifier, revision):
    """Lock the claim; refuse unless owner verifier and revision both match.

    An accepted completion closes the claim: no later phase may be written.
    """
    row = await fetch(conn, expected, lock=True)
    if (
        row is None
        or not secrets.compare_digest(row["owner"], verifier)
        or row["revision"] != revision
    ):
        seal.refuse()
    if await conn.fetchval(
        f"SELECT 1 FROM {COMPLETIONS} WHERE tenant_id=$1 AND invocation_id=$2",
        expected["tenant"], expected["invocation"],
    ):
        seal.refuse()
    return row