Handling Errors in Tasks¶

When a task raises an exception, worker catches the exception, marks the task failed, and stores it on the task as a TaskException:

@task
def divide(x: int, y: int) -> float:
    return x / y

# after a worker processes divide(1, 0):
# task.status == 'failed'
# task.exception.type == 'ZeroDivisionError'
# task.exception.message == 'division by zero'
# str(task.exception) == "ZeroDivisionError: division by zero"

TaskException also carries the exception's module, sanitized args, and the full formatted traceback, so a failure can be inspected from another process. Calling task.exception.to_exception() rebuilds the original exception on a best-effort basis. When the class cannot be imported or constructed, you get a TaskFailedError instead as a fallback.

Retries¶

Set retry to re-run a failed task automatically, and retry_delay to wait between attempts:

@task(retry=3, retry_delay=[1, 10, 60])
async def charge_card(order_id: int) -> str:
    ...

• A single number (retry_delay=5.0) waits the same time before every attempt.
• A list sets the delay per attempt: 1s before the first retry, 10s before the second, 60s before the third. When attempts outnumber list entries, the last value is reused.
• Between attempts the task has status retrying; it becomes failed only when no attempts remain. task.retry_count tracks how many retries have happened.

Retry a task manually with queue.retry():

await queue.retry(task)                           # re-queue now
await queue.retry(task, at=timedelta(hours=1))    # re-queue later
await queue.retry(task, force=True)               # re-run even a completed task

Or from the CLI: sheppy task retry <task-id>.

Timeouts¶

timeout caps execution time in seconds. A task that exceeds it fails with a TaskTimeoutError:

@task(timeout=30)
async def generate_report(report_id: int) -> str:
    ...

Timed-out tasks are not retried by default: a task that hangs once usually hangs again. Set retry_on_timeout=True to opt in.

Worker crashes¶

If a worker process dies mid-task, another worker eventually reclaims the task and marks it crashed, with a synthesized WorkerCrashedError exception (no traceback, since nothing was raised). For consistency reasons, crashed tasks are not retried by default. Use retry_on_crash=True if you want to automatically rerun the task on worker crash:

@task(retry=3, retry_on_crash=True)
async def import_data(file_path: str) -> str:
    ...

How long failures are kept¶

Finished tasks, including failed ones, keep their metadata in the backend until it expires. Failed and crashed tasks use error_ttl if set, otherwise the normal ttl. Set either to None to keep records forever.

Backend defaults are resolved when a task is enqueued and stay in its serialized configuration. Explicit task settings such as @task(ttl=300) always take precedence. Workflow steps inherit the workflow's resolved defaults, nested workflows inherit from their parent, and chained tasks inherit from the parent task. Older serialized tasks that still contain "inherit" fall back to the storing backend's defaults. See the TaskConfig reference for ttl/error_ttl semantics.

Example: a task that fails once¶

examples/testing/test_retry_logic.py defines a task that fails on its first attempt and succeeds on retry, and asserts on both attempts:

from sheppy import TestQueue, task

FAIL_COUNTER = 0

@task(retry=2, retry_delay=0)
def fail_once() -> str:
    global FAIL_COUNTER

    if FAIL_COUNTER < 1:
        FAIL_COUNTER += 1
        raise ValueError("task failed")

    return "success"

def test_fail_once():
    q = TestQueue()

    # instantiate the task
    t = fail_once()

    # add the task to the test queue
    q.add(t)

    assert q.size() == 1  # one task in the queue

    # process all tasks in the queue
    processed = q.process_all()

    # there should be two processed tasks: the original + one retry
    assert len(processed) == 2

    # verify the first attempt failed and is marked for retry
    assert processed[0].status == 'retrying'
    assert str(processed[0].exception) == "ValueError: task failed"

    # retry should succeed
    assert processed[1].status == 'completed'
    assert processed[1].exception is None
    assert processed[1].result == "success"

    # both processed tasks should have the same id
    assert processed[0].id == processed[1].id

Note the first processed attempt has status retrying, not failed. The task is failed only after the last attempt.