symfonic.memory.backends.postgres_erase_vector_chunks

postgres_erase_vector_chunks

Durable vector keyset chunks for permanent native tenant jobs.

The caller independently proves physical absence after the final chunk before retiring the permanent job and writing its receipt in the same transaction.

advance async

advance(conn: Any, jobs: str, job: Any, scope: Any) -> bool

Erase one transactional chunk; return True when no rows remain in range.

The caller holds the job row FOR UPDATE and commits the cursor together with the delete. A crash before commit restores both; a retry seeks after it.

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

async def advance(conn: Any, jobs: str, job: Any, scope: Any) -> bool:
    """Erase one transactional chunk; return True when no rows remain in range.

    The caller holds the job row FOR UPDATE and commits the cursor together with
    the delete. A crash before commit restores both; a retry seeks after it.
    """
    from symfonic.memory.backends.postgres_erase_vector_proof import require_ready
    from symfonic.memory.backends.postgres_native_subtree import _descends

    if not await installed(conn, jobs):
        raise MemoryContractError(UNREADY)
    schema = await conn.fetchval(
        "SELECT n.nspname FROM pg_class j JOIN pg_namespace n ON n.oid=j.relnamespace "
        "WHERE j.oid=$1::regclass", jobs,
    )
    table = await require_ready(conn, schema)
    tenant, prefix = job["tenant_id"], job["prefix"]
    bounds = (tenant, prefix, prefix[:-1] + "0")
    cursor = job["content_scope"] is not None
    if cursor != (job["content_key"] is not None):
        raise MemoryContractError("native vector erase cursor incomplete")
    after = (job["content_scope"], job["content_key"]) if cursor else ()
    await conn.execute(INCREMENTAL_SORT_OFF)
    async with index_probes(conn, ordered=True, presorted=True):
        rows = await conn.fetch(keyset_sql(table, cursor=cursor), *bounds, *after)
        if any(not _descends(row["scope_path"], scope) for row in rows):
            raise MemoryContractError("native retirement residual tenant data: memory_vectors")
        if rows:
            await _erase(conn, table, bounds, rows)
    if len(rows) < CHUNK_BUDGET:
        return True
    status = await conn.execute(
        f"UPDATE {jobs} SET content_scope=$2, content_key=$3 "
        "WHERE id=$1 AND state='erasing' AND phase='vectors'",
        job["id"], rows[-1]["scope_path"], rows[-1]["id"],
    )
    if status != "UPDATE 1":
        raise MemoryContractError("native vector erase cursor changed")
    return False

installed async

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

Attest the opt-in phase marker, irreversible guard and exact keyset index.

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

async def installed(conn: Any, jobs: str) -> bool:
    """Attest the opt-in phase marker, irreversible guard and exact keyset index."""
    from symfonic.memory.backends import postgres_erase_vector_subtree as subtree

    subtree_ready = await subtree.installed(conn, jobs)
    present = await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='vector_phase_marker' AND NOT attisdropped)", jobs,
    )
    if not present:
        return False
    if not await conn.fetchval(
        "SELECT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid=$1::regclass "
        "AND attname='vector_phase_marker' AND atttypid='boolean'::regtype "
        "AND attnotnull AND NOT attisdropped) AND EXISTS "
        "(SELECT 1 FROM pg_constraint WHERE conrelid=$1::regclass "
        "AND conname='memory_erase_jobs_vector_phase_marker' AND convalidated "
        "AND pg_get_constraintdef(oid)='CHECK (vector_phase_marker)') "
        "AND EXISTS (SELECT 1 FROM pg_constraint WHERE conrelid=$1::regclass "
        "AND conname='memory_erase_jobs_phase_known' AND convalidated "
        "AND pg_get_constraintdef(oid)=$4) "
        "AND EXISTS (SELECT 1 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=$2 AND t.tgenabled='A' AND t.tgtype=23 "
        "AND NOT t.tgisinternal AND t.tgqual IS NULL AND t.tgattr=''::int2vector "
        "AND p.pronamespace=j.relnamespace AND p.proowner=j.relowner "
        "AND p.prolang=(SELECT oid FROM pg_language WHERE lanname='plpgsql') "
        "AND p.prorettype='trigger'::regtype AND p.pronargs=0 "
        "AND p.provolatile='v' AND p.proconfig IS NULL "
        "AND NOT p.prosecdef AND p.prosrc=ANY($3::text[]))",
        jobs, GUARD, [subtree.VECTOR_BODY] if subtree_ready else [GUARD_BODY],
        PHASE_DEF,
    ):
        raise MemoryContractError(UNREADY)
    schema = jobs.rsplit(".", 1)[0]
    vector = f'{schema}."memory_vectors"'
    if not await conn.fetchval(_INDEX, vector, KEYSET):
        raise MemoryContractError(UNREADY)
    return True