Read at most budget+1 pending candidates inside the caller's fence.
The partial pending index keeps committed history out of the scan; invalid
owners count toward the budget and are parsed out before hydration. Falsey
markers and retained MVCC versions still cost physical work, bounded only
by the deadline. Index validation and selection share one deadline
and transaction; refusal precedes any flush/discard mutation. This helper
grants no public backend admission or destructive lifecycle deadline.
Source code in src/symfonic/memory/backends/postgres_pending.py
| async def read_pending(pool: Any, scope: NativeStorageScope) -> list[Any]:
"""Read at most budget+1 pending candidates inside the caller's fence.
The partial pending index keeps committed history out of the scan; invalid
owners count toward the budget and are parsed out before hydration. Falsey
markers and retained MVCC versions still cost physical work, bounded only
by the deadline. Index validation and selection share one deadline
and transaction; refusal precedes any flush/discard mutation. This helper
grants no public backend admission or destructive lifecycle deadline.
"""
from symfonic.memory.backends.postgres_graph import row_to_node
conn = await pool.acquire()
try:
async with native_subtree_deadline(conn):
await require_pending_index(conn)
jobs = await jobs_table(conn)
prefix = scope.identity.subtree_prefix
rows = await conn.fetch(
_held_sql(jobs), scope.tenant_id, prefix, prefix[:-1] + "0", PENDING_KEY,
)
finally:
await pool.release(conn)
if len(rows) > NATIVE_SUBTREE_BUDGET:
raise MemoryContractError("native subtree budget exceeded: pending rows")
if any(row["held"] for row in rows):
raise MemoryContractError(ERASING)
root = scope.identity.levels
nodes = []
for row in rows:
try:
levels = StorageScopeKey.parse(row["scope_path"]).levels
except MemoryContractError:
continue
if len(levels) < len(root) or levels[:len(root)] != root:
continue
node = row_to_node(row)
if is_pending(node.properties):
nodes.append(node)
from symfonic.memory.backends.postgres_staged_fence import refuse_claimed
# Before any flush/discard mutation: only the exact selected publication
# may release receipt-claimed rows, staged or in flight.
refuse_claimed(nodes)
return nodes
|