Fix SoftTimeLimitExceeded: celery.exceptions.SoftTimeLimitExceeded: SoftTimeLimitExceeded() in Celery

This error occurs when a Celery task exceeds its soft time limit. Celery raises SoftTimeLimitExceeded as an exception inside the task so you can perform cleanup. Fix it by catching the exception to save partial progress, optimizing the task to run faster, breaking it into smaller subtasks, or increasing the time limit if the work genuinely takes longer.

Reading the Stack Trace

Here's what each line means:

• File "/app/app/tasks.py", line 24, in generate_report: The task was executing a heavy database query when the soft time limit was reached.
• File "/app/venv/lib/python3.12/site-packages/celery/app/trace.py", line 477, in trace_task: Celery's task tracer caught the SoftTimeLimitExceeded exception and marked the task as failed.
• celery.exceptions.SoftTimeLimitExceeded: SoftTimeLimitExceeded(): The task exceeded the configured soft_time_limit, giving it a chance to clean up before being killed.

Common Causes

@celery.task(soft_time_limit=300)
def generate_report(start_date, end_date):
    rows = db.session.execute(text('SELECT * FROM orders')).fetchall()  # millions of rows
    return process_all(rows)

@celery.task(soft_time_limit=60)
def send_bulk_emails(user_ids):
    for uid in user_ids:
        send_email(uid)  # no try/except, progress lost if limit hit

@celery.task(soft_time_limit=10)  # 10 seconds for a heavy report
def generate_report():
    ...

The Fix

Break the work into batches and catch SoftTimeLimitExceeded to save progress and reschedule remaining work. This ensures partial results are not lost and the task can resume from where it left off. Increase the soft_time_limit to a more realistic value for each batch.

@celery.task(soft_time_limit=300)
def generate_report(start_date, end_date):
    rows = db.session.execute(text('SELECT * FROM orders')).fetchall()
    return process_all(rows)

from celery.exceptions import SoftTimeLimitExceeded

@celery.task(bind=True, soft_time_limit=600, time_limit=660)
def generate_report(self, start_date, end_date, offset=0, batch_size=1000):
    try:
        rows = db.session.execute(
            text('SELECT * FROM orders LIMIT :limit OFFSET :offset'),
            {'limit': batch_size, 'offset': offset}
        ).fetchall()
        result = process_batch(rows)
        if len(rows) == batch_size:
            # Schedule next batch
            generate_report.delay(start_date, end_date, offset + batch_size, batch_size)
        return result
    except SoftTimeLimitExceeded:
        # Save progress and reschedule remaining work
        save_partial_result(offset)
        generate_report.delay(start_date, end_date, offset, batch_size)
        return {'status': 'partial', 'offset': offset}

Testing the Fix

import pytest
from unittest.mock import patch, MagicMock
from celery.exceptions import SoftTimeLimitExceeded
from app.tasks import generate_report

@pytest.fixture
def celery_app():
    from app import create_app
    app = create_app()
    app.config['CELERY_ALWAYS_EAGER'] = True
    return app

def test_report_processes_batch(celery_app):
    with celery_app.app_context():
        result = generate_report('2026-01-01', '2026-03-31', offset=0, batch_size=10)
        assert result is not None

@patch('app.tasks.db.session.execute')
def test_soft_time_limit_saves_progress(mock_execute, celery_app):
    mock_execute.side_effect = SoftTimeLimitExceeded()
    with celery_app.app_context():
        result = generate_report('2026-01-01', '2026-03-31')
        assert result['status'] == 'partial'

def test_batch_schedules_next_chunk(celery_app):
    with celery_app.app_context():
        with patch('app.tasks.generate_report.delay') as mock_delay:
            generate_report('2026-01-01', '2026-03-31', offset=0, batch_size=1)
            # If there are more rows, delay should be called
            # This depends on test data availability

Run your tests:

pytest tests/ -v

Pushing Through CI/CD

git checkout -b fix/celery-task-timeout-batching,git add app/tasks.py tests/test_tasks.py,git commit -m "fix: batch report generation and handle SoftTimeLimitExceeded",git push origin fix/celery-task-timeout-batching

Your CI config should look something like this:

name: CI
on:
  pull_request:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: 'pip'
      - run: pip install -r requirements.txt
      - run: pytest tests/ -v --tb=short
        env:
          CELERY_BROKER_URL: redis://localhost:6379/0

The Full Manual Process: 18 Steps

Here's every step you just went through to fix this one bug:

1. Notice the error alert or see it in your monitoring tool
2. Open the error dashboard and read the stack trace
3. Identify the file and line number from the stack trace
4. Open your IDE and navigate to the file
5. Read the surrounding code to understand context
6. Reproduce the error locally
7. Identify the root cause
8. Write the fix
9. Run the test suite locally
10. Fix any failing tests
11. Write new tests covering the edge case
12. Run the full test suite again
13. Create a new git branch
14. Commit and push your changes
15. Open a pull request
16. Wait for code review
17. Merge and deploy to production
18. Monitor production to confirm the error is resolved

Total time: 30-60 minutes. For one bug.

pip install bugstack

import bugstack

bugstack.init(api_key=os.environ["BUGSTACK_API_KEY"])

Deploying the Fix (Manual Path)

1. Run pytest locally to confirm batching and timeout handling work correctly.
2. Open a pull request with the task refactor and tests.
3. Wait for CI checks to pass on the PR.
4. Have a teammate review and approve the PR.
5. Merge to main and monitor task durations and completion rates in staging.

Frequently Asked Questions