UNDF: UNDF-2026-000000025

UNDF: (pending)

celery-MOADX-0001: BaseBackend.get_task_meta — stampede O(N) concurrent backend fetches on shared result cache

MOAD-0006 Candidate — The Thundering Herd

Field	Value
ID	celery-MOADX-0001
Severity	HIGH
Ecosystem	Celery
Language	Python
File	`celery/backends/base.py`
Lines	722–745
Pattern	LRUCache[key] → KeyError → expensive backend fetch → LRUCache[key]=meta (no lock around fetch)
Trigger	N threads simultaneously poll the same task_id before result is cached; `result_backend_thread_safe=True`
Compute cost	Network round-trip to Redis / database / AMQP broker

Defect

BaseBackend.get_task_meta is the result-retrieval path called by every AsyncResult.get() call. When the Celery app is configured with result_backend_thread_safe = True, the backend instance is stored as an app-level singleton (_backend_cache) shared across all threads:

# celery/app/base.py:1468-1469
if backend.thread_safe:
    self._backend_cache = backend  # shared singleton

The shared backend contains self._cache, a kombu.utils.functional.LRUCache (backed by threading.RLock) that caches successful task results:

# celery/backends/base.py:722-743
def get_task_meta(self, task_id, cache=True):
    self._ensure_not_eager()
    if cache:
        try:
            return self._cache[task_id]     # <-- RLock, returns if hit
        except KeyError:
            pass

    # NO LOCK AROUND THIS BLOCK:
    meta = self._ensure_retryable(
        self._get_task_meta_for,            # <-- network call to Redis/DB
        fallback_exc=BackendGetMetaError,
        fallback_msg="failed to get meta",
        task_id=task_id
    )

    if cache and meta.get('status') == states.SUCCESS:
        self._cache[task_id] = meta         # <-- RLock
    return meta

The LRUCache.__getitem__ and __setitem__ each hold self.mutex (RLock) individually, but the three-step sequence — check for hit → fetch from backend → write to cache — is not atomic.

Stampede scenario: A task completes. At the moment of completion, N threads are all blocked in AsyncResult.get() polling the same task_id. The task result is not yet in _cache:

All N threads call get_task_meta(task_id).
All N get KeyError from self._cache[task_id].
All N call self._get_task_meta_for(task_id) — each sends an independent query to Redis/database.
All N receive the same result and write it to self._cache[task_id] (redundant).

Grower signal: Works fine at low concurrency. With N=100 threads all waiting on the same long-running task (common in fan-out → fan-in patterns with group() + chord()), result publication triggers 100 simultaneous Redis GET commands instead of 1.

The pattern also applies to get_group_meta (lines 769–775) which follows an identical structure for group-level result caching.

Fix

Option A — per-key threading.Lock with double-checked locking (preferred):

import threading

class BaseBackend:
    def __init__(self, ...):
        ...
        self._cache = _nulldict() if cmax == -1 else LRUCache(limit=cmax)
        self._pending_meta = {}          # key → threading.Event
        self._pending_meta_lock = threading.Lock()

    def get_task_meta(self, task_id, cache=True):
        self._ensure_not_eager()
        if cache:
            try:
                return self._cache[task_id]
            except KeyError:
                pass

        # Singleflight: only one thread fetches, others wait
        with self._pending_meta_lock:
            if task_id in self._pending_meta:
                event = self._pending_meta[task_id]
                is_worker = False
            else:
                event = threading.Event()
                self._pending_meta[task_id] = event
                is_worker = True

        if not is_worker:
            event.wait(timeout=30)
            try:
                return self._cache[task_id]
            except KeyError:
                pass
            # fallthrough: worker failed, retry ourselves

        try:
            meta = self._ensure_retryable(
                self._get_task_meta_for,
                fallback_exc=BackendGetMetaError,
                fallback_msg="failed to get meta",
                task_id=task_id
            )
            if cache and meta.get('status') == states.SUCCESS:
                self._cache[task_id] = meta
            return meta
        finally:
            if is_worker:
                with self._pending_meta_lock:
                    self._pending_meta.pop(task_id, None)
                event.set()

Option B — accept redundant fetches, add comment (low-cost mitigation):

Not recommended. This only documents the defect without fixing it.

Option C — use result_backend_thread_safe = False (default) to avoid sharing:

Avoids the defect by giving each thread its own backend instance, but consumes more memory and connections. Not a fix — a workaround.

The same pattern applies to get_group_meta and should receive the same treatment.

Impact

Redis stampede: N simultaneous GET commands per result-ready task when N threads poll the same task_id. At N=100 threads waiting on a chord callback, result publication triggers 100 Redis round-trips instead of 1.
Database stampede: With database backends (Django ORM, SQLAlchemy), N concurrent SELECT queries hit the same row.
Fix: After fix, first thread fetches, all others wait on Event and read from cache. Network round-trips per task result: 1 regardless of fan-in depth.

5.6 KiB Raw Permalink Blame History Unescape Escape