symfonic.memory.backends.postgres_erase_retirement

postgres_erase_retirement

Explicit terminal tenant retirement on the native erase-job authority (CORE-426).

The operator must migrate and re-install eraser grants before using this API. Older fleets must be drained: pre-CORE-426 readers only understand erasing.

installed async

installed(conn: Any, jobs: str) -> bool

An old job schema remains valid; a partial retirement install refuses.

Source code in src/symfonic/memory/backends/postgres_erase_retirement.py

async def installed(conn: Any, jobs: str) -> bool:
    """An old job schema remains valid; a partial retirement install refuses."""
    present = await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='permanent' AND NOT attisdropped)",
        jobs,
    )
    if not present:
        if not await conn.fetchval(
            "SELECT EXISTS (SELECT 1 FROM pg_constraint WHERE conrelid=$1::regclass "
            "AND contype='c' AND convalidated AND pg_get_constraintdef(oid)=$2)",
            jobs,
            "CHECK ((state = 'erasing'::text))",
        ):
            raise MemoryContractError(UNMIGRATED)
        return False
    if not await conn.fetchval(
        "SELECT (SELECT count(*)=3 FROM pg_constraint WHERE conrelid=$1::regclass "
        "AND contype='c' AND convalidated AND pg_get_constraintdef(oid)=ANY($2::text[])) "
        "AND EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='permanent' AND atttypid='boolean'::regtype AND attnotnull)",
        jobs,
        list(CONSTRAINT_DEFS),
    ):
        raise MemoryContractError(UNMIGRATED)
    vector_phase = await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='vector_phase_marker' AND NOT attisdropped)",
        jobs,
    )
    guard_body = (
        GUARD_BODY.replace("'claims'", "'vectors'").replace("claims phase", "vectors phase")
        if vector_phase
        else GUARD_BODY
    )
    guarded = await conn.fetchval(
        "SELECT count(*)=2 FROM pg_trigger t JOIN pg_proc p ON p.oid=t.tgfoid "
        "JOIN pg_class j ON j.oid=t.tgrelid WHERE j.oid=$1::regclass "
        "AND t.tgname=ANY($2::text[]) AND t.tgenabled='A' AND NOT t.tgisinternal "
        "AND t.tgqual IS NULL AND t.tgattr=''::int2vector "
        "AND ((t.tgname=$2[1] AND t.tgtype=31) OR (t.tgname=$2[2] AND t.tgtype=34)) "
        "AND p.pronamespace=j.relnamespace AND p.proowner=j.relowner "
        "AND p.prosrc=$3 AND NOT p.prosecdef AND j.relpersistence='p' "
        "AND NOT j.relrowsecurity",
        jobs,
        [GUARD, GUARD + "_truncate"],
        guard_body,
    )
    if not guarded:
        raise MemoryContractError(UNMIGRATED)
    if not await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_index i JOIN pg_class x ON x.oid=i.indexrelid "
        "JOIN pg_am m ON m.oid=x.relam JOIN pg_attribute a "
        "ON a.attrelid=i.indrelid AND a.attname='tenant_id' "
        "WHERE i.indrelid=$1::regclass AND i.indisvalid AND i.indisready "
        "AND i.indisunique AND i.indnkeyatts=1 AND i.indkey[0]=a.attnum "
        "AND m.amname='btree' AND i.indcollation[0]=a.attcollation "
        "AND pg_get_expr(i.indpred,i.indrelid)='permanent')",
        jobs,
    ):
        raise MemoryContractError(UNMIGRATED)
    return True

preflight async

preflight(conn: Any, jobs: str, scope: Any) -> None

Refuse unerasable ownership before creating an irreversible permanent job.

Source code in src/symfonic/memory/backends/postgres_erase_retirement.py

async def preflight(conn: Any, jobs: str, scope: Any) -> None:
    """Refuse unerasable ownership before creating an irreversible permanent job."""
    from symfonic.memory.backends.postgres_erase_ownership import preflight as indexed

    await indexed(conn, jobs, scope)

refuse_admission async

refuse_admission(conn: Any, tenant: str) -> None

Retirement blocks fresh turns without changing ordinary forget admission.

Source code in src/symfonic/memory/backends/postgres_erase_retirement.py

async def refuse_admission(conn: Any, tenant: str) -> None:
    """Retirement blocks fresh turns without changing ordinary forget admission."""
    from symfonic.memory.backends.postgres_erase_jobs import jobs_table

    jobs = await jobs_table(conn)
    if jobs is not None and await conn.fetchval(
        f"SELECT EXISTS (SELECT 1 FROM {jobs} j WHERE tenant_id=$1 "
        "AND (state='retired' OR to_jsonb(j)->>'permanent'='true'))",
        tenant,
    ):
        raise MemoryContractError(RETIRED)

require_empty async

require_empty(conn: Any, jobs: str, job_id: str) -> None

Never certify hidden legacy, malformed, or unscoped tenant-owned rows.

Source code in src/symfonic/memory/backends/postgres_erase_retirement.py

async def require_empty(conn: Any, jobs: str, job_id: str) -> None:
    """Never certify hidden legacy, malformed, or unscoped tenant-owned rows."""
    from symfonic.memory.backends.postgres_erase_ownership import require_empty as indexed

    await indexed(conn, jobs, job_id)

retire async

retire(eraser: Any, scope: Any, *, runtime_login: str, binding: Any) -> Any

Retire a tenant root; repeated retirement verifies completion and its anchor.

Source code in src/symfonic/memory/backends/postgres_erase_retirement.py

async def retire(eraser: Any, scope: Any, *, runtime_login: str, binding: Any) -> Any:
    """Retire a tenant root; repeated retirement verifies completion and its anchor."""
    from symfonic.capabilities.memory.scope import MemoryScope
    from symfonic.memory.backends.postgres_erase_content import content_ready
    from symfonic.memory.backends.postgres_erase_create import _record
    from symfonic.memory.backends.postgres_erase_routing import (
        OVERLAPPING,
        overlap_args,
        overlap_sql,
    )
    from symfonic.memory.backends.postgres_memory_generation import advance
    from symfonic.memory.backends.postgres_native_eraser import _LIFECYCLE_LOCK, EraseStep
    from symfonic.memory.backends.postgres_native_subtree import GRAPH_LOCK
    from symfonic.memory.storage_scope import NativeStorageScope

    root = NativeStorageScope.from_scope(MemoryScope.from_levels((("tenant", scope.tenant_id),)))
    if scope.subtree_prefix != root.subtree_prefix:
        raise MemoryContractError("native retirement requires the tenant root")
    await eraser._verified()
    completed = False
    async with eraser._transaction(binding) as conn:
        jobs = await eraser._jobs(conn, runtime_login)
        if not await installed(conn, jobs) or not await content_ready(conn, jobs):
            raise MemoryContractError(UNMIGRATED)
        await conn.execute(_LIFECYCLE_LOCK, f"symfonic-memory:{scope.tenant_id}")
        rows = await conn.fetch(
            f"SELECT id, prefix, state FROM {jobs} WHERE tenant_id=$1 FOR UPDATE",
            scope.tenant_id,
        )
        # Chunks lock the job before pinning proof indexes; preserve that order.
        await conn.execute(GRAPH_LOCK)
        terminal = next((row for row in rows if row["state"] == "retired"), None)
        if terminal is not None:
            job_id, completed = terminal["id"], True
            await require_empty(conn, jobs, job_id)
            receipts = jobs.rsplit(".", 1)[0] + ".memory_erase_receipts"
            from symfonic.memory.backends.postgres_erase_receipts import prefix_digest
            from symfonic.memory.backends.postgres_retirement_receipt import verify

            if eraser._receipt_keys is not None:
                await verify(conn, jobs, receipts, job_id, eraser)
            elif not await conn.fetchval(
                f"SELECT EXISTS (SELECT 1 FROM {receipts} WHERE id=$1 AND prefix_digest=$2)",
                job_id,
                prefix_digest(eraser._receipt_key, scope.tenant_id, scope.subtree_prefix),
            ):
                raise MemoryContractError("native retirement receipt missing or invalid")
        else:
            await preflight(conn, jobs, scope)
            overlaps = await conn.fetch(overlap_sql(jobs), *overlap_args(scope))
            if overlaps and (len(overlaps) != 1 or overlaps[0]["prefix"] != scope.subtree_prefix):
                raise MemoryContractError(OVERLAPPING)
            from symfonic.memory.backends.postgres_retirement_receipt import initiation

            receipt_identity = await initiation(conn, jobs, eraser)
            if overlaps:
                job_id = overlaps[0]["id"]
                if receipt_identity is not None and not await conn.fetchval(
                    f"SELECT receipt_key_id IS NOT NULL FROM {jobs} WHERE id=$1",
                    job_id,
                ):
                    raise MemoryContractError("unversioned erase job cannot gain receipt identity")
            else:
                job_id = await _record(
                    conn,
                    jobs,
                    scope,
                    "edges",
                    receipt_identity=receipt_identity,
                )
            promoted = await conn.fetchval(
                f"UPDATE {jobs} SET permanent=true WHERE id=$1 AND NOT permanent RETURNING id",
                job_id,
            )
            schema = jobs.rsplit(".", 1)[0]
            from symfonic.memory.backends import postgres_journal_base_retirement as journal
            from symfonic.memory.backends.postgres_erase_ownership import staged_installed

            if await staged_installed(conn, schema):
                from symfonic.memory.backends import postgres_staged_auxiliary_retirement as staged

                await staged.erase(conn, schema, scope)
            await journal.erase(conn, schema, scope)
            from symfonic.memory.backends.postgres_auxiliary_retirement import erase

            await erase(conn, schema, scope)
            if promoted is not None:
                await advance(conn, scope)
    if completed:
        await eraser.proof.acknowledge(eraser.pool)
        return EraseStep(job_id, True)
    try:
        return await eraser._advance(job_id, binding)
    except MemoryContractError as error:
        if str(error) != RETIRED:
            raise
        # Another retirement caller completed between our commit and chunk.
        return await retire(eraser, scope, runtime_login=runtime_login, binding=binding)