symfonic.memory.backends.postgres_erase_jobs

postgres_erase_jobs

Durable native subtree erase jobs and the writer fence they impose (CORE-199 S1).

A large forget erases what one transaction cannot in later chunks: claim identities, and content past its budget (postgres_erase_content). Between those commits a job row marks the subtree as erasing: no native mutation may touch it, no claim reader may treat it as live, and every content read hides it (erasing, live_paths). This module ships the table, the fence and the hiding predicate; nothing here creates a job.

The fence is sound only after the writer holds a memory_nodes lock in the same transaction. The sweep that will commit a job takes EXCLUSIVE on memory_nodes, which conflicts with every lock a writer takes, so either the writer committed first and the sweep erases its rows, or the writer waited and its next READ COMMITTED statement sees the committed job. A check made before the lock, or under an older snapshot, misses a job committed in between, so other isolation levels refuse while the table exists.

The jobs that can fence a writer are the ones beside the memory_nodes it locked, so the table is resolved in that relation's schema and every query names it schema-qualified. A job table reachable only elsewhere on search_path, one the session cannot read, or one without the canonical prefix constraint refuses: each would otherwise skip the fence or read the wrong schema.

A prefix must be a complete-level canonical v1 key rooted at the row's own tenant. On such a prefix the writer's exact-ancestor test and the reader's starts_with agree; without the final slash a reader would also fence a sibling whose ID extends the job's, and the writer would not.

canonical_job_prefix

canonical_job_prefix(tenant_id: str, prefix: str) -> bool

Python mirror of PREFIX_CHECK: S2 validates a job before inserting it.

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

def canonical_job_prefix(tenant_id: str, prefix: str) -> bool:
    """Python mirror of PREFIX_CHECK: S2 validates a job before inserting it."""
    if type(prefix) is not str or not re.fullmatch(_PREFIX_BODY, prefix):
        return False
    root = prefix[_HEADER_LENGTH:].split("/", 1)[0]
    return root.split(":")[1] == tenant_id.encode("utf-8").hex()

fence_scope async

fence_scope(conn: Any, scope: Any) -> None

For a writer about to touch rows owned at scope or above it.

ROW SHARE on memory_nodes first, then the fence, inside the writer's own transaction. Once content erases in chunks (CORE-199), an update or a delete of a row an erasing job still holds would act on content the forget already answered for, so these writers refuse like inserts do. A v0 scope owns no canonical key, so no job can fence it.

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

async def fence_scope(conn: Any, scope: Any) -> None:
    """For a writer about to touch rows owned at ``scope`` or above it.

    ROW SHARE on memory_nodes first, then the fence, inside the writer's own
    transaction. Once content erases in chunks (CORE-199), an update or a
    delete of a row an erasing job still holds would act on content the
    forget already answered for, so these writers refuse like inserts do.
    A v0 scope owns no canonical key, so no job can fence it.
    """
    from symfonic.memory.scope_isolation import _is_native_scope

    if not _is_native_scope(scope):
        return
    await share_lock_nodes(conn)
    await refuse_if_erasing(conn, scope.tenant_id, (scope,))

grant_read_only async

grant_read_only(conn: Any, jobs: str, role: str) -> None

Owner side, in the caller's transaction: reset role to SELECT on jobs.

role must already be a validated plain identifier.

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

async def grant_read_only(conn: Any, jobs: str, role: str) -> None:
    """Owner side, in the caller's transaction: reset ``role`` to SELECT on ``jobs``.

    ``role`` must already be a validated plain identifier.
    """
    await conn.execute(f'REVOKE ALL ON {jobs} FROM "{role}"')
    await conn.execute(f'GRANT SELECT ON {jobs} TO "{role}"')
    if drift := await read_only_drift(conn, jobs, role):
        raise MemoryContractError(drift)

jobs_table async

jobs_table(conn: Any, schema: str | None = None) -> str | None

The schema-qualified job table, or None; readers need no isolation guard.

schema resolves beside schema.memory_nodes for a reader whose search_path is pinned away from it.

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

async def jobs_table(conn: Any, schema: str | None = None) -> str | None:
    """The schema-qualified job table, or None; readers need no isolation guard.

    ``schema`` resolves beside ``schema.memory_nodes`` for a reader whose
    search_path is pinned away from it.
    """
    jobs, _ = await _resolve(conn, schema)
    return jobs

read_only_drift async

read_only_drift(conn: Any, jobs: str, role: str) -> str | None

How role departs from SELECT-only on jobs, or None if it does not.

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

async def read_only_drift(conn: Any, jobs: str, role: str) -> str | None:
    """How ``role`` departs from SELECT-only on ``jobs``, or None if it does not."""
    row = await conn.fetchrow(_HELD, role, jobs, list(_EXCESS))
    if not row["reads"]:
        return f"{role} cannot SELECT the erase job table {jobs}"
    if row["excess"]:
        return f"{role} holds {row['excess'][0]} on the erase job table {jobs}"
    return None

refuse_absent_if_erasing async

refuse_absent_if_erasing(pool: Any, scope: Any) -> None

Read-only: a record a job hides is held, not absent (CORE-199).

For a caller that found nothing at scope and is about to report so: reads hide what an erasing job holds, so the answer there is the erase in progress, as a write would get. Nothing is mutated either way.

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

async def refuse_absent_if_erasing(pool: Any, scope: Any) -> None:
    """Read-only: a record a job hides is held, not absent (CORE-199).

    For a caller that found nothing at ``scope`` and is about to report so:
    reads hide what an erasing job holds, so the answer there is the erase
    in progress, as a write would get. Nothing is mutated either way.
    """
    from symfonic.memory.scope_isolation import _is_native_scope

    if not _is_native_scope(scope):
        return
    conn = await pool.acquire()
    try:
        jobs = await jobs_table(conn)
        if jobs is not None and await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM {jobs} WHERE tenant_id=$1 "
            "AND prefix=ANY($2::text[]) AND state IN ('erasing','retired'))",
            scope.tenant_id, sorted(set(scope.ancestor_prefix_paths())),
        ):
            raise MemoryContractError(ERASING)
    finally:
        await pool.release(conn)

refuse_if_erasing async

refuse_if_erasing(conn: Any, tenant_id: str, scopes: Iterable[Any]) -> None

Refuse a native mutation whose scope lies in an erasing subtree.

Call only after this transaction holds its memory_nodes lock. A job at an ancestor key or at the key itself fences a write; a job below the write's own scope does not, since the write lands outside that subtree.

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

async def refuse_if_erasing(conn: Any, tenant_id: str, scopes: Iterable[Any]) -> None:
    """Refuse a native mutation whose scope lies in an erasing subtree.

    Call only after this transaction holds its memory_nodes lock. A job at an
    ancestor key or at the key itself fences a write; a job below the write's
    own scope does not, since the write lands outside that subtree.
    """
    prefixes = sorted({key for scope in scopes for key in scope.ancestor_prefix_paths()})
    if not prefixes:
        return
    jobs, isolation = await _resolve(conn)
    if jobs is None:
        return
    if isolation != "read committed":
        raise MemoryContractError("native erase fence requires READ COMMITTED isolation")
    if await conn.fetchval(
        f"SELECT EXISTS (SELECT 1 FROM {jobs} "
        "WHERE tenant_id=$1 AND prefix=ANY($2::text[]) AND state IN ('erasing','retired'))",
        tenant_id,
        prefixes,
    ):
        raise MemoryContractError(ERASING)

share_lock_nodes async

share_lock_nodes(conn: Any) -> None

For a writer that locks no graph row: ROW SHARE on memory_nodes, first.

ROW SHARE is the weakest mode the sweep's EXCLUSIVE conflicts with, so a job committed with the sweep is visible to this transaction's later refuse_if_erasing. Taken before the writer's own rows, it orders memory_nodes first, as every forget does, so neither waits on the other in a cycle. PostgreSQL grants this mode only to a role with INSERT, UPDATE, DELETE or TRUNCATE on memory_nodes; any other refuses here rather than write unfenced.

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

async def share_lock_nodes(conn: Any) -> None:
    """For a writer that locks no graph row: ROW SHARE on memory_nodes, first.

    ROW SHARE is the weakest mode the sweep's EXCLUSIVE conflicts with, so a
    job committed with the sweep is visible to this transaction's later
    ``refuse_if_erasing``. Taken before the writer's own rows, it orders
    memory_nodes first, as every forget does, so neither waits on the other
    in a cycle. PostgreSQL grants this mode only to a role with INSERT,
    UPDATE, DELETE or TRUNCATE on memory_nodes; any other refuses here
    rather than write unfenced.
    """
    jobs, _ = await _resolve(conn)
    if jobs is None:
        return
    try:
        await conn.execute(f"LOCK TABLE {jobs.rsplit('.', 1)[0]}.memory_nodes IN ROW SHARE MODE")
    except Exception as error:
        if getattr(error, "sqlstate", None) == "42501":
            raise MemoryContractError(UNLOCKABLE) from error
        raise