bq wraps every processor call in db.begin_nested() (a SAVEPOINT) on the session it hands you. Two consequences for processors that wait on long subprocesses (e.g. a 20-minute eval/agent run):
db.commit() inside a processor breaks both prod (commits inside bq's begin_nested context manager -> 'Can't operate on closed transaction inside context manager') and connection-bound test fixtures.idle_in_transaction_session_timeout (if configured on the engine): the connection is killed mid-processor, every later session use raises PendingRollbackError, and bq requeues the task forever — re-running paid work each cycle.Working pattern (three moves):
proc.wait(timeout=30) cycles) and tick a keepalive each cycle: db.execute(select(1)) resets the idle-in-transaction timer, so the strict leak detector can stay ON.with Session(db.get_bind(), expire_on_commit=False) as work_db: ...; work_db.commit(). In prod that's a fresh pooled connection with real commits; under a connection-bound pytest fixture (external connection.begin() + Session(bind=connection)) SQLAlchemy 2.0's default join_transaction_mode='conditional_savepoint' turns the inner commit into a savepoint release, so test isolation is preserved with NO harness changes.Also add pool_pre_ping=True to the worker engine so a dead pooled connection is replaced at checkout instead of exploding the next task.