symfonic.memory.backends.postgres_pending_claims

postgres_pending_claims

Durable identity for private native PostgreSQL pending writes.

An accepted token survives publication in this ledger. A later retry may therefore prove its own earlier commit even after pending row markers vanish. Unclaimed pre-v11 rows are deliberately not adopted by guessing from an ID.

erase_subtree async

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

Forget all claim identities in the graph sweep's same transaction.

Published generations stay in the ledger, so a subtree's claims grow with rewrites while its live rows do not. Read at most budget+1 tokens through the owner C index as index probes and refuse above the budget; the caller's transaction then rolls the whole forget back. Every native claim insert follows its node upsert in one transaction, so the sweep's EXCLUSIVE node lock closes this set; a claim still found after the token delete refuses rather than survive a forget reported complete. The delete must remove every token read, each still inside the owner range: a claim moved or removed by another writer in between refuses, and the rollback restores a moved claim where that writer put it.

The budget counts visible claims. Dead index entries of erased claims are still visited until vacuum removes them, and only the forget deadline bounds that read.

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

async def erase_subtree(conn: Any, scope: Any) -> None:
    """Forget all claim identities in the graph sweep's same transaction.

    Published generations stay in the ledger, so a subtree's claims grow with
    rewrites while its live rows do not. Read at most budget+1 tokens through
    the owner C index as index probes and refuse above the budget; the caller's
    transaction then rolls the whole forget back. Every native claim insert
    follows its node upsert in one transaction, so the sweep's EXCLUSIVE node
    lock closes this set; a claim still found after the token delete refuses
    rather than survive a forget reported complete. The delete must remove
    every token read, each still inside the owner range: a claim moved or
    removed by another writer in between refuses, and the rollback restores
    a moved claim where that writer put it.

    The budget counts visible claims. Dead index entries of erased claims are
    still visited until vacuum removes them, and only the forget deadline
    bounds that read.
    """
    from symfonic.memory.backends.postgres_index_probes import index_probes
    from symfonic.memory.backends.postgres_native_subtree_query import _require_c_range_index

    budget = CLAIM_ERASE_BUDGET
    bounds, in_range = _bounds(scope), _IN_RANGE
    await _require_c_range_index(conn, "memory_pending_claims", _CLAIM_SCOPE_INDEX)
    async with index_probes(conn, ordered=True):
        rows = await conn.fetch(
            f"SELECT token FROM memory_pending_claims WHERE {in_range} "
            'ORDER BY tenant_id COLLATE "C", scope_path COLLATE "C" '
            f"LIMIT {budget + 1}", *bounds,
        )
        if len(rows) > budget:
            raise MemoryContractError("native subtree budget exceeded: claim history")
        # The delete names tokens only, so the primary key is its one index
        # path. With the range in its WHERE, a generic plan estimated the
        # range at a few rows, scanned it and matched each row against the
        # whole token array: a full budget missed the deadline. The range is
        # rechecked on the erased rows instead.
        erased, outside = await conn.fetchrow(
            "WITH erased AS (DELETE FROM memory_pending_claims WHERE token=ANY($4::text[]) "
            f"RETURNING tenant_id, scope_path) SELECT count(*), count(*) FILTER "
            f"(WHERE NOT ({in_range})) FROM erased",
            *bounds, [row["token"] for row in rows],
        )
        if erased != len(rows) or outside or await conn.fetchval(
            f"SELECT EXISTS (SELECT 1 FROM memory_pending_claims WHERE {in_range})", *bounds,
        ):
            raise MemoryContractError("native subtree claim history changed during erase")

over_budget async

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

Whether the subtree holds more claims than one forget may erase inline.

Asked before any delete, under the sweep's EXCLUSIVE node lock, so a forget that must become a durable job refuses or hands off before it mutates anything. The read stops at budget+1 claims.

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

async def over_budget(conn: Any, scope: Any) -> bool:
    """Whether the subtree holds more claims than one forget may erase inline.

    Asked before any delete, under the sweep's EXCLUSIVE node lock, so a
    forget that must become a durable job refuses or hands off before it
    mutates anything. The read stops at budget+1 claims.
    """
    from symfonic.memory.backends.postgres_index_probes import index_probes
    from symfonic.memory.backends.postgres_native_subtree_query import _require_c_range_index

    await _require_c_range_index(conn, "memory_pending_claims", _CLAIM_SCOPE_INDEX)
    async with index_probes(conn, ordered=True):
        return await conn.fetchval(
            f"SELECT count(*) > {CLAIM_ERASE_BUDGET} FROM (SELECT 1 FROM memory_pending_claims "
            f'WHERE {_IN_RANGE} ORDER BY tenant_id COLLATE "C", scope_path COLLATE "C" '
            f"LIMIT {CLAIM_ERASE_BUDGET + 1}) s", *_bounds(scope),
        )

publish_node async

publish_node(pool: Any, scope: Any, node: Any) -> None

Publish only the exact pending version and mark its claim atomically.

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

async def publish_node(pool: Any, scope: Any, node: Any) -> None:
    """Publish only the exact pending version and mark its claim atomically."""
    from symfonic.memory.backends.postgres_erase_jobs import refuse_if_erasing
    from symfonic.memory.pending import pending_jsonb_condition, published_properties

    token = token_of(node.properties)
    properties = {
        **published_properties(node.properties),
        "__spreading_access_count": int(node.spreading_access_count),
    }
    conn = await pool.acquire()
    try:
        async with conn.transaction():
            row = await conn.fetchrow(
                'UPDATE memory_nodes SET properties=$5::jsonb, updated_at=NOW() '
                'WHERE tenant_id=$1 AND id=$2 AND scope_path COLLATE "C"=$3 '
                'AND tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 '
                'AND properties->>$7=$4 AND '
                + pending_jsonb_condition("properties", 6)
                + ' RETURNING id', scope.tenant_id, str(node.id), scope.scope_path,
                token, json.dumps(properties), PENDING_KEY, PENDING_BATCH_KEY,
            )
            await refuse_if_erasing(conn, scope.tenant_id, (scope,))
            if row is None:
                raise MemoryContractError("native pending claim changed before publication")
            await mark(conn, scope, str(node.id), token, "published")
    finally:
        await pool.release(conn)

state_of async

state_of(pool: Any, scope: Any, record_id: str, token: str) -> str | None

A claim under an erasing subtree refuses rather than read as any state.

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

async def state_of(pool: Any, scope: Any, record_id: str, token: str) -> str | None:
    """A claim under an erasing subtree refuses rather than read as any state."""
    from symfonic.memory.backends.postgres_erase_jobs import ERASING, fenced_claim, jobs_table

    conn = await pool.acquire()
    try:
        jobs = await jobs_table(conn)
        fenced = jobs is not None
        row = await conn.fetchrow(
            _STATE_OF.format(f", {fenced_claim(jobs)} AS fenced" if fenced else ""),
            token, scope.tenant_id, record_id, scope.scope_path,
        )
    finally:
        await pool.release(conn)
    if row is not None and fenced and row["fenced"]:
        raise MemoryContractError(ERASING)
    return None if row is None else row["state"]

states async

states(pool: Any, claims: tuple[tuple[str, str, str], ...]) -> set[str]

Return only IDs whose exact owner and token were published.

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

async def states(pool: Any, claims: tuple[tuple[str, str, str], ...]) -> set[str]:
    """Return only IDs whose exact owner and token were published."""
    if not claims:
        return set()
    from symfonic.memory.backends.postgres_erase_jobs import ERASING, fenced_claim, jobs_table

    conn = await pool.acquire()
    try:
        jobs = await jobs_table(conn)
        fenced = jobs is not None
        rows = await conn.fetch(
            "SELECT token,tenant_id,record_id,scope_path"
            + (f", {fenced_claim(jobs)} AS fenced" if fenced else "")
            + " FROM memory_pending_claims c "
            "WHERE token=ANY($1::text[]) AND state='published'",
            [token for _, _, token in claims],
        )
    finally:
        await pool.release(conn)
    from symfonic.capabilities.memory.scope import scope_from_path
    from symfonic.memory.storage_scope import NativeStorageScope

    known = {
        (
            scope_from_path(path).tenant, record_id,
            NativeStorageScope.from_scope(scope_from_path(path)).scope_path, token,
        )
        for path, record_id, token in claims
    }
    owned = [
        row for row in rows
        if (row["tenant_id"], row["record_id"], row["scope_path"], row["token"])
        in known
    ]
    # A published claim whose content an erase job already removed is absent,
    # and a retry must not reconcile against it as an earlier commit.
    if fenced and any(row["fenced"] for row in owned):
        raise MemoryContractError(ERASING)
    return {row["record_id"] for row in owned}