Long-running BeanQueue (bq) task processor doing network/LLM work in a loop intermittently fails with sqlalchemy InvalidRequestError "Can't operate on closed transaction" and psycopg2 QueryCanceled. Stack frames point inside the processor's scrape loop and at the processor function itself. Non-obvious contract violations discovered by reading bq source: (1) bq's Processor.process wraps the processor call in with db.begin_nested() on the SAME session that loaded the task row and that later records task state — so the injected db has a transaction open from task-SELECT until bq's post-task commit, for the processor's entire runtime; (2) any db.rollback() inside the processor (e.g. in a per-item except handler) rolls back past bq's savepoint, so the begin_nested() context exit and/or the next ORM op raises "Can't operate on closed transaction"; (3) if the engine sets idle_in_transaction_session_timeout via connect_args, the task session sits idle-in-transaction for the whole task while the processor does network/LLM work, and Postgres kills the backend before bq's final task-state commit; (4) early flush() of a status column (e.g. last_attempt_at) takes a row lock held across the entire loop, so a second long transaction updating the same rows blocks and hits statement_timeout → QueryCanceled.
Treat the bq-injected session as bq's property: never commit or roll it back, and don't do long-running work on it. Fix pattern: (a) materialize the IDs/keys you need from the injected session up front; (b) run each loop iteration on a dedicated short-lived session — with Session(db.get_bind(), expire_on_commit=False) as work_db: — and let the worker function commit at checkpoints (immediately after writing status/lock-taking columns, after each merge) so row locks are held for milliseconds, not the whole scrape; (c) in per-item except handlers, roll back only the dedicated session (or just let the context manager close it), never the injected one; (d) once per loop iteration, ping the injected session with db.execute(select(1)) — each statement resets Postgres's idle_in_transaction_session_timeout timer, so bq's final task-state commit survives multi-hour tasks. Bonus testability property: Session(bind=outer_session.get_bind()) does the right thing in both environments — in production the bind is an Engine (fresh pooled connection, independent commits); under the standard join-external-transaction test fixture the bind is a Connection with an open transaction, and SQLAlchemy 2.0's default join_transaction_mode='conditional_savepoint' makes the inner session's commit() a RELEASE SAVEPOINT, preserving per-test rollback isolation (verified semantics on SQLAlchemy 2.0.46).