# 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:

```python
# 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:

```python
# 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`:

1. All N threads call `get_task_meta(task_id)`.
2. All N get `KeyError` from `self._cache[task_id]`.
3. All N call `self._get_task_meta_for(task_id)` — each sends an independent query to Redis/database.
4. 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):**

```python
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.
