symfonic.memory.backends.postgres_forget_deadline

postgres_forget_deadline

Private graph forget budget and explicit commit outcome handling.

ForgetCommitIndeterminateError

Bases: RuntimeError

COMMIT was not confirmed; reconcile durable state before retrying.

This does not establish rollback. PostgreSQL can complete COMMIT after the client loses its acknowledgement, including deferred trigger work which PostgreSQL 16 does not cancel with statement_timeout.

forget_transaction async

forget_transaction(pool: Any, *, deadline_at: float | None = None)

Bound graph forget and distinguish confirmed from uncertain commit.

The ambient binding makes graph operations join the fence's connection. A savepoint protects an existing caller transaction on refusal. LOCAL settings remain effective through top-level COMMIT. In a caller transaction, restoration happens before RELEASE so refusal can still roll back the deletion. Disposal after a successful COMMIT cannot report budget refusal. Acquisition, locks and mutations share the one-second deadline. COMMIT is shielded for a 250ms acknowledgement grace after expiry; an unconfirmed outcome discards the connection and raises a distinct error. Any isolation but READ COMMITTED, own or inherited, refuses before work.

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

@asynccontextmanager
async def forget_transaction(pool: Any, *, deadline_at: float | None = None):
    """Bound graph forget and distinguish confirmed from uncertain commit.

    The ambient binding makes graph operations join the fence's connection.
    A savepoint protects an existing caller transaction on refusal. LOCAL
    settings remain effective through top-level COMMIT. In a caller transaction,
    restoration happens before RELEASE so refusal can still roll back the
    deletion. Disposal after a successful COMMIT cannot report budget refusal.
    Acquisition, locks and mutations share the one-second deadline. COMMIT is
    shielded for a 250ms acknowledgement grace after expiry; an
    unconfirmed outcome discards the connection and raises a distinct error.
    Any isolation but READ COMMITTED, own or inherited, refuses before work.
    """
    if foreign_bound_transaction(pool):
        raise RuntimeError("inherited PostgreSQL transaction belongs to another task")
    existing = bound_connection(pool)
    if _ACTIVE.get() is pool and existing is not None:
        yield existing
        return
    conn = binding = active = None
    if deadline_at is None:
        deadline_at = asyncio.get_running_loop().time() + NATIVE_SUBTREE_TIMEOUT_MS / 1000
    try:
        try:
            async with asyncio.timeout_at(deadline_at) as deadline:
                conn = await pool.acquire()
                if existing is None:
                    owner = asyncio.current_task()
                    if owner is None:
                        raise RuntimeError("PostgreSQL forget requires an asyncio task")
                    binding = _TRANSACTION.set((pool, conn, owner))
                active = _ACTIVE.set(pool)
                previous = await conn.fetchval("SHOW statement_timeout")
                nested = conn.is_in_transaction()
                transaction = conn.transaction()
                await transaction.start()
                try:
                    # A snapshot older than the EXCLUSIVE locks hides rows
                    # committed while the sweep waited, so the completeness
                    # checks would certify a forget that left them behind. A
                    # savepoint inherits its caller's level; SHOW takes no
                    # snapshot, so this refuses before snapshot or lock.
                    if await conn.fetchval("SHOW transaction_isolation") != "read committed":
                        raise MemoryContractError(
                            "native subtree forget requires READ COMMITTED isolation"
                        )
                    await conn.fetchval(
                        "SELECT set_config('statement_timeout', "
                        "(LEAST(NULLIF(setting::int, 0), $1))::text || 'ms', true) "
                        "FROM pg_settings WHERE name='statement_timeout'",
                        NATIVE_SUBTREE_TIMEOUT_MS,
                    )
                    yield conn
                    if nested:
                        await conn.fetchval(
                            "SELECT set_config('statement_timeout', $1, true)", previous,
                        )
                    else:
                        remaining = max(1, int(
                            (deadline.when() - asyncio.get_running_loop().time()) * 1000,
                        ))
                        await conn.fetchval(
                            "SELECT set_config('statement_timeout', "
                            "(LEAST(NULLIF(setting::int, 0), $1))::text || 'ms', true) "
                            "FROM pg_settings WHERE name='statement_timeout'", remaining,
                        )
                    # A nested operation may catch a SQL error while leaving
                    # the transaction aborted. PostgreSQL then answers COMMIT
                    # with a ROLLBACK tag that asyncpg's transaction API hides.
                    await conn.fetchval("SELECT 1")
                except BaseException:
                    await transaction.rollback()
                    raise
                else:
                    await _confirmed_commit(transaction, conn, deadline)
        except Exception as exc:
            if isinstance(exc, TimeoutError) or getattr(exc, "sqlstate", None) == "57014":
                raise MemoryContractError("native subtree execution time budget exceeded") from exc
            raise
    finally:
        if active is not None:
            _ACTIVE.reset(active)
        if binding is not None:
            _TRANSACTION.reset(binding)
        if conn is not None:
            await pool.release(conn)