Skip to content

symfonic.memory.backends.postgres_working_expiry

postgres_working_expiry

Explicit offline immutable expiry authority; no automatic TTL enablement.

install_working_expiry async 

install_working_expiry(installer_pool, guard, *, fleet_attestation=None)

Versioned additive offline migration over an attested v1 guard.

Locks nodes/edges/guard against writers. Existing registrations backfill as permanent; TTL was refused by the old host. Never invoke during admission. An explicit operator fleet attestation is required; it is not drain proof.

Source code in src/symfonic/memory/backends/postgres_working_expiry.py 
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
65
66
67
68
69
70
71
72
async def install_working_expiry(installer_pool, guard, *, fleet_attestation=None):
    """Versioned additive offline migration over an attested v1 guard.

    Locks nodes/edges/guard against writers. Existing registrations backfill as
    permanent; TTL was refused by the old host. Never invoke during admission.
    An explicit operator fleet attestation is required; it is not drain proof.
    """
    from symfonic.memory.backends.postgres_working_expiry_fleet import (
        require_expiry_fleet_attestation,
    )
    require_expiry_fleet_attestation(guard, fleet_attestation)
    if bound_connection(installer_pool) is not None:
        raise RuntimeError('Working expiry installation requires its own transaction')
    conn = await installer_pool.acquire()
    try:
        async with conn.transaction():
            await conn.execute("SET LOCAL lock_timeout = '5s'")
            await conn.execute(
                "SELECT pg_advisory_xact_lock("
                "hashtextextended('symfonic-working-guard-install',0))"
            )
            if await conn.fetchval('SELECT current_schema()') != guard.schema:
                raise RuntimeError('Working expiry installation requires the guard schema')
            if await conn.fetchval('SELECT current_user') == guard.working_host_role:
                raise RuntimeError('Working expiry installation requires the installer role')
            await attest_working_guard(
                conn, guard.schema, guard.quoted_schema,
                WORKING_GUARD_DDL.replace('__GUARD_SCHEMA__', guard.quoted_schema),
                guard.working_host_role,
            )
            await conn.execute(
                f'LOCK TABLE {guard.quoted_schema}.memory_nodes, '
                f'{guard.quoted_schema}.memory_edges, '
                f'{guard.quoted_schema}.memory_working_endpoint_guard '
                'IN SHARE ROW EXCLUSIVE MODE'
            )
            exists = await conn.fetchval('SELECT to_regclass($1) IS NOT NULL',
                guard.quoted_schema + '.memory_working_expiry')
            if not exists:
                marker = await conn.fetchval(
                    "SELECT col_description(a.attrelid,a.attnum) FROM pg_attribute a "
                    "WHERE a.attrelid=to_regclass($1) AND a.attname='retired'",
                    guard.quoted_schema + '.memory_working_endpoint_guard',
                )
                if marker is not None:
                    raise RuntimeError('Working expiry migration is incomplete or conflicts')
                await conn.execute(EXPIRY_DDL.replace('__SCHEMA__', guard.quoted_schema)
                                   .replace('__HOST__', guard.working_host_role))
                await conn.execute(
                    f'COMMENT ON COLUMN {guard.quoted_schema}.'
                    'memory_working_endpoint_guard.retired '
                    f"IS 'symfonic-working-expiry-v1:{guard.working_host_role}'"
                )
            await attest_working_expiry(conn, guard)
    finally:
        await installer_pool.release(conn)

register_working_expiry async 

register_working_expiry(guard, scope, record_id, ttl_hours=None)

Register expiry once, in the host's guarded node-creation transaction.

Source code in src/symfonic/memory/backends/postgres_working_expiry.py 
75
76
77
78
79
80
81
82
async def register_working_expiry(guard, scope, record_id, ttl_hours=None):
    """Register expiry once, in the host's guarded node-creation transaction."""
    conn = await _transaction_connection(guard)
    tenant, owner = _v0_owner(scope)
    await conn.execute(
        f'SELECT {guard.quoted_schema}.symfonic_register_working_expiry($1,$2,$3,$4)',
        tenant, record_id, owner, ttl_hours,
    )