Skip to content

symfonic.memory.backends.postgres_working_expiry_attestation

postgres_working_expiry_attestation

One catalog statement per read; exact TTL authority checks without a cache.

discover_working_expiry async 

discover_working_expiry(conn, *, schema=None, host_role=None)

One fresh catalog snapshot: absent v0 or fully attested expiry authority.

Source code in src/symfonic/memory/backends/postgres_working_expiry_attestation.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
async def discover_working_expiry(conn, *, schema=None, host_role=None):
    """One fresh catalog snapshot: absent v0 or fully attested expiry authority."""
    row = await conn.fetchrow(
        _CATALOG, schema, host_role, json.dumps(_BODIES), json.dumps(_COLUMNS),
    )
    if row is None:
        raise RuntimeError('Working expiry requires a trusted current schema')
    if row['failure'] is not None:
        raise RuntimeError('Working expiry catalog attestation failed: ' + row['failure'])
    if not row['installed']:
        return None
    quoted = '"' + row['nspname'].replace('"', '""') + '"'
    return quoted + '.symfonic_working_expiry_visible'