symfonic.memory.backends.postgres_candidate_budget

postgres_candidate_budget

Physical row budget for native PostgreSQL candidate selection (CORE-199).

Ranking scores every eligible row (token hits, ancestor decay, the profile slot) before LIMIT, so a cue-bearing read must see the whole eligible set. The native read therefore materializes at most CANDIDATE_SCAN_BUDGET + 1 rows and refuses when the set is larger. It never truncates recall and never reports a partial page as complete; under budget it ranks exactly the rows the unbounded statement ranked.

With idx_nodes_candidate_eligible_c installed the budget counts eligible rows only (postgres_candidate_index). Without it, it counts every visible row in the owner paths, whatever its layer or eligibility, because only the C range index condition bounds what that scan reads. Dead index entries and TTL-expired rows are still visited without counting; the native read deadline bounds that residual.

Empty cue: every row of one (path, layer, audience, profile) seek shares its ancestor decay and cue factor, so the score orders exactly as the index rank key, provided the decay power(0.5, d), a numeric that the float8 rank product casts implicitly, is the exact power of two 2**-d. Then the global top limit (plus the profile slot) is drawn from each seek's first limit rows, and one more row per seek decides completeness: a page never holds more than limit rows of one seek, profile slots included. Rows too wide for the eligible key have no rank order, so their seeks read every such row; the page draws from a superset and stays exact. Every fetched row counts against the budget, so an empty-cue read stays exact only while (limit + 1) times the populated seeks, plus the wide rows, fits it; past that it refuses rather than truncates. PostgreSQL 16's numeric power stops being exact at d = 23, so deeper scopes keep the counting budget.

bounded_source

bounded_source(paths_index: int, paths_sql: str | None = None) -> str

CTEs reading at most budget + 1 visible rows, and the refusal flag.

paths_sql replaces the owner array (CORE-199 erasing owners removed).

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

def bounded_source(paths_index: int, paths_sql: str | None = None) -> str:
    """CTEs reading at most budget + 1 visible rows, and the refusal flag.

    ``paths_sql`` replaces the owner array (CORE-199 erasing owners removed).
    """
    paths = paths_sql or f"${paths_index}::text[]"
    return f"""visible AS MATERIALIZED (
 SELECT * FROM memory_nodes WHERE tenant_id COLLATE "C"=$1
 AND scope_path COLLATE "C"=ANY({paths}) LIMIT {_limit()}
), {_flag()}"""

eligible_source

eligible_source(seeks: str) -> str

CTEs keeping at most budget + 1 eligible seek rows, and the refusal flag.

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

def eligible_source(seeks: str) -> str:
    """CTEs keeping at most budget + 1 eligible seek rows, and the refusal flag."""
    return f"""{seeks} visible AS MATERIALIZED (SELECT * FROM seeks LIMIT {_limit()}), {_flag()}"""

ordered_index_scan async

ordered_index_scan(conn: Any)

Pin the bounded read to an ordered index scan; restore on every exit.

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

@asynccontextmanager
async def ordered_index_scan(conn: Any):
    """Pin the bounded read to an ordered index scan; restore on every exit."""
    previous = await conn.fetchval(
        "SELECT json_object_agg(name, setting) FROM (SELECT name, setting, "
        "set_config(name, 'off', true) FROM pg_settings WHERE name=ANY($1::text[])) s",
        list(_PINNED),
    )
    restore = "SELECT set_config(key, value, true) FROM json_each_text($1::json)"
    try:
        yield
    except BaseException:
        # An aborted transaction refuses the restore and has discarded the
        # local settings anyway; the original error is the one to surface.
        with suppress(Exception):
            await conn.execute(restore, previous)
        raise
    await conn.execute(restore, previous)

ranked_rows

ranked_rows(rows: list[Any], counted: str = 'visible') -> list[Any]

Refuse an over-budget row set; drop the flag-only row of an empty page.

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

def ranked_rows(rows: list[Any], counted: str = "visible") -> list[Any]:
    """Refuse an over-budget row set; drop the flag-only row of an empty page."""
    if rows and rows[0]["budget_exceeded"]:
        from symfonic.capabilities.memory.errors import MemoryContractError

        raise MemoryContractError(
            f"native candidate scan budget exceeded: more than {CANDIDATE_SCAN_BUDGET} "
            f"{counted} rows"
        )
    return [row for row in rows if row["id"] is not None]

seek_cap

seek_cap(scope: Any, request: Any, tokens: list[str]) -> int

Rows per seek: the budget itself unless the empty-cue top-k is exact.

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

def seek_cap(scope: Any, request: Any, tokens: list[str]) -> int:
    """Rows per seek: the budget itself unless the empty-cue top-k is exact."""
    if tokens or len(scope.ancestor_prefix_paths()) > EXACT_DECAY_LEVELS:
        return CANDIDATE_SCAN_BUDGET + 1
    return request.limit + 1

visible_paths

visible_paths(scope: Any, request: Any) -> list[str]

Owner paths the ranked statement may read: ancestors, admitted, exact.

Equality under COLLATE "C" is byte equality, so the intersection is the same set the SQL predicates select; one array keeps it an index condition.

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

def visible_paths(scope: Any, request: Any) -> list[str]:
    """Owner paths the ranked statement may read: ancestors, admitted, exact.

    Equality under COLLATE "C" is byte equality, so the intersection is the
    same set the SQL predicates select; one array keeps it an index condition.
    """
    paths = scope.ancestor_prefix_paths()
    if request.allowed_scope_paths is not None:
        allowed = set(request.allowed_scope_paths)
        paths = [path for path in paths if path in allowed]
    if request.exact_scope:
        paths = [path for path in paths if path == materialise_scope_path(scope)]
    return paths