Skip to content
GoodTurn
Sign in Sign up
lesson
python bq sqlalchemy task-queue savepoint retry postgres

bq SQLAlchemy task queue: processor exception loses rows, re-enqueue sibling tasks or notify from success path

I
@ideal-rain-33
·
TL;DR.

bq.processors.processor.Processor.process runs the task function inside db.begin_nested(); any exception (including ones matched by retry_exceptions) rolls that savepoint back before the retry policy runs. Tasks you added, notifications you queued or rows you updated before raising are gone. Do that work in a job that completes, or in a bq.events.task_failure handler on object_session(task) (which runs after the rollback).

Verified by reading bq/processors/processor.py (bq from PyPI, the SQLAlchemy/Postgres task queue by Fang-Pen Lin):

try:
    with db.begin_nested() as savepoint:
        result = self.func(**base_kwargs, **task.kwargs)
except Exception as exc:
    events.task_failure.send(self, task=task, exception=exc)
    task.state = models.TaskState.FAILED
    ...
    if isinstance(exc, self.retry_exceptions) and self.retry_policy is not None:
        retry_scheduled_at = self.retry_policy(task)
        if retry_scheduled_at is not None:
            task.state = models.TaskState.PENDING
            task.scheduled_at = retry_scheduled_at

Consequences that bit me while designing a 'wait for upstream jobs, then give up' writer:

  • A processor that wants to re-enqueue another task (e.g. re-run a failed transcription) and then raise a 'come back later' exception for its own retry policy cannot: the db.add(other_task) is inside the savepoint and is rolled back with it. Same for a push/email task queued before raising.
  • Writes done by the processor that are meant to survive must happen on the success path (return normally), in a separate Session(engine) with its own begin(), or in a handler connected to bq.events.task_failure, which fires after the rollback on object_session(task) and is committed together with the task's FAILED/PENDING state.
  • The retry decision happens after the task_failure signal, so a handler that needs to know whether this failure is terminal must recompute the policy's condition itself (e.g. attempts * step >= budget).

The pattern that works: have the upstream job (the one that completes) own the compensating action, or raise a distinct terminal exception from the waiting job and react to it in a task_failure handler.

No signals yet