symfonic.memory.backends.postgres_erase_chunks

postgres_erase_chunks

One keyset chunk of a durable native erase job (CORE-199 S3).

A claims-phase job's graph content is already gone: either it left with the job's first transaction, or content chunks (postgres_erase_content) erased it before the job reached this phase. A permanent job with the opt-in vector phase continues to vector chunks after claims. Each chunk is its own transaction on the eraser's connection and erases at most CHUNK_BUDGET claims:

• SELECT ... FOR UPDATE on the job row serializes resumers. A second resumer waits, then re-reads the row as the first left it: the advanced cursor, or no row once the first completed the job. No chunk is erased twice and none is skipped.
• The read seeks (scope_path, token) after the persisted cursor through the keyset index, so dead entries earlier chunks left behind are never revisited, however old a snapshot keeps them from being pruned.
• The delete names tokens only and re-checks the range on what it erased; any claim that moved or vanished in between refuses and rolls the chunk back, cursor included.
• The cursor advances in the same commit. A crash after commit resumes from it; a crash before commit leaves cursor and claims as they were.

A chunk that reads fewer than the budget has reached the end of the range. Ordinary jobs delete the job and write a digest-only receipt in that commit. An opted-in permanent job moves to vector chunks and stays fenced permanently. Completion independently proves physical absence before certifying its receipt. No new claim can enter the range behind the cursor: every native writer and claim reader refuses under an erasing job (S1), checked after its memory_nodes lock, and the job committed under EXCLUSIVE, before or together with the content erase. A writer still running pre-S1 code is the exception, and only the operator's fleet barrier excludes it.

Chunk transactions take no table lock beyond ROW EXCLUSIVE on the claims, so another tenant's writes never wait on them.

advance async

advance(conn: Any, jobs: str, receipts: str, claims: str, job_id: str, key: bytes) -> bool | None

Erase one chunk; True once complete, False while more remain, None if unknown.

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

async def advance(
    conn: Any, jobs: str, receipts: str, claims: str, job_id: str, key: bytes
) -> bool | None:
    """Erase one chunk; True once complete, False while more remain, None if unknown."""
    done, _ = await step(conn, jobs, receipts, claims, job_id, key)
    return done

keyset_sql

keyset_sql(claims: str, *, cursor: bool) -> str

The chunk read; tests EXPLAIN this exact text.

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

def keyset_sql(claims: str, *, cursor: bool) -> str:
    """The chunk read; tests EXPLAIN this exact text."""
    return (
        f"SELECT scope_path, token FROM {claims} WHERE {_RANGE}"
        f"{_AFTER if cursor else ''} {_ORDER} LIMIT {CHUNK_BUDGET}"
    )

step async

step(conn: Any, jobs: str, receipts: str, claims: str, job_id: str, key: bytes, receipt_authority: Any = None) -> tuple[bool | None, str | None]

advance, plus the phase the chunk ran in (None for an unknown job).

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

async def step(
    conn: Any,
    jobs: str,
    receipts: str,
    claims: str,
    job_id: str,
    key: bytes,
    receipt_authority: Any = None,
) -> tuple[bool | None, str | None]:
    """``advance``, plus the phase the chunk ran in (None for an unknown job)."""
    from symfonic.memory.backends.postgres_index_probes import index_probes

    job = await conn.fetchrow(_JOB.format(jobs=jobs), job_id)
    if job is None:
        from symfonic.memory.backends.postgres_erase_retirement import RETIRED

        if await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {jobs} WHERE id=$1 AND state='retired')",
            job_id,
        ):
            raise MemoryContractError(RETIRED)
        done = await conn.fetchval(f"SELECT EXISTS (SELECT 1 FROM {receipts} WHERE id=$1)", job_id)
        return (True if done else None), None
    phase = job["phase"] or "claims"
    receipt_key = None
    if job["receipt_key_id"] is not None:
        from symfonic.memory.backends.postgres_retirement_receipt import require_identity

        receipt_key = await require_identity(conn, job, receipt_authority)
    if phase == "vectors":
        from symfonic.capabilities.memory.storage_key import StorageScopeKey
        from symfonic.memory.backends import postgres_erase_vector_chunks as vectors
        from symfonic.memory.backends.postgres_erase_ownership import require_indexes
        from symfonic.memory.storage_scope import NativeStorageScope

        # Completion needs graph proof locks as well as the vector relation.
        # Pin graph first, matching retirement admission's GRAPH_LOCK order.
        await require_indexes(conn, jobs.rsplit(".", 1)[0])
        scope = NativeStorageScope(StorageScopeKey.parse(job["prefix"]))
        if not await vectors.advance(conn, jobs, job, scope):
            return False, phase
        if await conn.fetchval(
            f"SELECT COALESCE((to_jsonb(j)->>'vector_forget')::boolean, false) "
            f"FROM {jobs} j WHERE id=$1",
            job_id,
        ):
            from symfonic.memory.backends.postgres_erase_vector_subtree import require_empty

            await require_empty(conn, jobs, scope)
        receipt_key_id = job["receipt_key_id"]
        sealed = prefix_digest(key, job["tenant_id"], job["prefix"])
        if receipt_key_id is not None:
            from symfonic.memory.backends.postgres_retirement_receipt import INVALID, digest

            if receipt_key is None:
                raise MemoryContractError(INVALID)
            sealed = digest(receipt_key, job)
        await _complete(
            conn,
            jobs,
            receipts,
            job_id,
            sealed,
            receipt_key_id,
        )
        return True, phase
    if phase != "claims":
        from symfonic.capabilities.memory.storage_key import StorageScopeKey
        from symfonic.memory.backends import postgres_erase_content as content
        from symfonic.memory.storage_scope import NativeStorageScope

        scope = NativeStorageScope(StorageScopeKey.parse(job["prefix"]))
        await content.advance(conn, jobs, job, scope)
        return False, phase
    tenant, prefix = job["tenant_id"], job["prefix"]
    bounds = (tenant, prefix, prefix[:-1] + "0")
    cursor = job["cursor_scope"] is not None
    after = (job["cursor_scope"], job["cursor_token"]) if cursor else ()
    async with index_probes(conn, ordered=True):
        rows = await conn.fetch(keyset_sql(claims, cursor=cursor), *bounds, *after)
        if rows:
            await _erase(conn, claims, bounds, [row["token"] for row in rows])
    if len(rows) < CHUNK_BUDGET:
        from symfonic.memory.backends.postgres_erase_vector_chunks import installed

        permanent = await conn.fetchval(
            f"SELECT COALESCE((to_jsonb(j)->>'permanent')::boolean, false) "
            f"FROM {jobs} j WHERE id=$1",
            job_id,
        )
        vector_forget = await conn.fetchval(
            f"SELECT COALESCE((to_jsonb(j)->>'vector_forget')::boolean, false) "
            f"FROM {jobs} j WHERE id=$1",
            job_id,
        )
        if (permanent or vector_forget) and await installed(conn, jobs):
            status = await conn.execute(
                f"UPDATE {jobs} SET phase='vectors', cursor_scope=NULL, "
                "cursor_token=NULL, content_scope=NULL, content_key=NULL "
                "WHERE id=$1 AND state='erasing' AND phase='claims'",
                job_id,
            )
            if status != "UPDATE 1":
                raise MemoryContractError("native vector erase transition changed")
            return False, phase
        receipt_key_id = job["receipt_key_id"]
        sealed = prefix_digest(key, tenant, prefix)
        if receipt_key_id is not None:
            from symfonic.memory.backends.postgres_retirement_receipt import INVALID, digest

            if receipt_key is None:
                raise MemoryContractError(INVALID)
            sealed = digest(receipt_key, job)
        await _complete(conn, jobs, receipts, job_id, sealed, receipt_key_id)
        return True, phase
    status = await conn.execute(
        f"UPDATE {jobs} SET cursor_scope=$2, cursor_token=$3 WHERE id=$1 AND state='erasing'",
        job_id,
        rows[-1]["scope_path"],
        rows[-1]["token"],
    )
    if status != "UPDATE 1":
        raise MemoryContractError("native erase job changed during its chunk")
    return False, phase