symfonic.memory.backends.postgres_erase_content

postgres_erase_content

Content erased in chunks by a durable native erase job (CORE-199).

A forget whose content (nodes, edges or workset progress) exceeds one transaction's budget becomes a job in phase edges: its first transaction, under the sweep's EXCLUSIVE graph lock, records the job and advances the subtree's erase generation and erases nothing else. From that commit on the whole subtree is gone to every reader, which hides rows under an erasing prefix in the same statement that reads them, and every native writer refuses under it (postgres_erase_jobs). So the forget is all or nothing to everyone but the eraser, and nothing can enter the range behind a cursor.

Each later chunk is one transaction on the eraser's login. It reads at most CONTENT_CHUNK keys of the current phase's table after the persisted cursor, in key order through a C keyset index, deletes the canonical descendants among them by primary key, re-checks that each deleted row was inside the range, and advances the cursor in the same commit. A chunk that reads fewer keys than the budget moves the job to the next phase: edges, then progress, then nodes, then claims, where the claim-history chunks (postgres_erase_chunks) finish it. Keys that do not parse as a descendant stay, exactly as an inline forget leaves them; no reader admits such a key.

Chunks take no table lock beyond ROW EXCLUSIVE, so writers of other tenants never wait on them. Without scripts/migrations/core199_erase_content.sql no content job can be created, and an oversized forget refuses as before.

advance async

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

Erase one chunk of the job's current phase; return the phase it leaves.

job is the job row the caller locked FOR UPDATE; scope is its subtree. Returns the phase the job is in after this chunk commits.

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

async def advance(conn: Any, jobs: str, job: Any, scope: Any) -> str:
    """Erase one chunk of the job's current phase; return the phase it leaves.

    ``job`` is the job row the caller locked FOR UPDATE; ``scope`` is its
    subtree. Returns the phase the job is in after this chunk commits.
    """
    from symfonic.memory.backends.postgres_index_probes import index_probes
    from symfonic.memory.backends.postgres_native_subtree import _descends

    phase = job["phase"]
    tenant, prefix = job["tenant_id"], job["prefix"]
    bounds = (tenant, prefix, prefix[:-1] + "0")
    cursor = job["content_scope"] is not None
    after = (job["content_scope"], job["content_key"]) if cursor else ()
    # The keyset index supplies the ORDER BY. With sorts off, incremental ones
    # included, no plan can trade it for the owner-range index plus a sort
    # that re-reads the cursor's owner. The chunk is its own transaction.
    await conn.execute(INCREMENTAL_SORT_OFF)
    async with index_probes(conn, ordered=True, presorted=True):
        rows = await conn.fetch(keyset_sql(phase, cursor=cursor), *bounds, *after)
        doomed = [row for row in rows if _descends(row["scope_path"], scope)]
        if doomed:
            await _erase(conn, phase, bounds, doomed)
    if len(rows) < CONTENT_CHUNK:
        following = PHASES[PHASES.index(phase) + 1]
        await _move(conn, jobs, job["id"], following, None, None)
        return following
    await _move(conn, jobs, job["id"], phase, rows[-1]["scope_path"], rows[-1]["key"])
    return phase

content_ready async

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

Whether the operator installed content jobs beside jobs.

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

async def content_ready(conn: Any, jobs: str) -> bool:
    """Whether the operator installed content jobs beside ``jobs``."""
    return bool(await conn.fetchval(_READY, jobs))

keyset_sql

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

One phase's chunk read; tests EXPLAIN this exact text.

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

def keyset_sql(phase: str, *, cursor: bool) -> str:
    """One phase's chunk read; tests EXPLAIN this exact text."""
    table, key = _TABLE[phase], _KEY[phase]
    if phase == "progress":
        # No tenant column: the prefix is rooted at the tenant (canonical key).
        where = '$1::text IS NOT NULL AND scope_path COLLATE "C">=$2 AND scope_path COLLATE "C"<$3'
        after = " AND (scope_path, cycle) > ($4::text, $5::text)"
        order = 'ORDER BY scope_path COLLATE "C", cycle'
    else:
        where = ('tenant_id COLLATE "C"=$1 AND scope_path COLLATE "C">=$2 '
                 'AND scope_path COLLATE "C"<$3')
        after = ' AND (scope_path COLLATE "C", id COLLATE "C") > ($4::text, $5::text)'
        order = 'ORDER BY tenant_id COLLATE "C", scope_path COLLATE "C", id COLLATE "C"'
    return (
        f"SELECT scope_path, {key} AS key FROM {table} WHERE {where}"
        f"{after if cursor else ''} {order} LIMIT {CONTENT_CHUNK}"
    )

over_budget async

over_budget(conn: Any, scope: Any) -> bool

Whether any content table holds more rows than one forget may erase.

Under the sweep's EXCLUSIVE lock, before any delete; each read stops at budget+1 entries of its C range index, as the inline selectors do.

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

async def over_budget(conn: Any, scope: Any) -> bool:
    """Whether any content table holds more rows than one forget may erase.

    Under the sweep's EXCLUSIVE lock, before any delete; each read stops at
    budget+1 entries of its C range index, as the inline selectors do.
    """
    from symfonic.memory.backends.postgres_native_subtree import _paths

    for table in ("memory_nodes", "memory_edges", "memory_workset_progress"):
        try:
            await _paths(conn, table, scope, bounded=True)
        except MemoryContractError as error:
            if str(error) != CONTENT_EXCEEDED:
                raise
            return True
    return False