Tuning Backpressure for Celery and RQ Catalog Workers
This page answers one narrow operational question: why does a burst of catalog updates make your worker fleet balloon in memory and start timing out tasks, even though the broker is healthy and the workers are “keeping up”? It sits under Async Batch Processing for Catalog Updates, which covers the distributed-ingestion contract end to end, and within the broader Catalog Ingestion & ILS Sync Pipelines architecture. If you run Celery for distributed catalog ingestion — or the equivalent on RQ — against a rate-limited ILS API or a single-writer database, this is the failure you hit the first time a vendor drops a 200,000-record overlay load on you.
Problem Framing
The symptom arrives in a rush. A vendor overlay job or a full reharvest enqueues tens of thousands of catalog-update tasks in seconds. The broker — Redis or RabbitMQ — accepts them instantly because a broker’s whole job is to absorb bursts cheaply. The workers dutifully drain the queue as fast as they can. Then the resident memory of each worker process climbs, the ILS REST API starts returning 429 and 503, task runtimes stretch from milliseconds to tens of seconds, and eventually the process is killed by the out-of-memory reaper or the soft time limit fires mid-write.
The tell is that the broker is fine while the workers are not. Sampling the queue and a worker together during a burst shows the mismatch directly:
$ celery -A catalog inspect stats | grep -E 'prefetch|rusage.maxrss'
"prefetch_count": 128,
"rusage": { "maxrss": 2113904 } # ~2.0 GiB RSS per worker
$ redis-cli LLEN catalog.updates
(integer) 48213
$ tail -n2 /var/log/catalog/worker.log
WARNING task catalog.apply_update timed out (soft_time_limit=30s), retrying
ERROR ils client: 429 Too Many Requests (Retry-After: 12) on PUT /v6/bibs/...
Prefetch is 128, so each worker has reserved 128 tasks — each carrying a full MARC payload — in memory before the first one is even written to the ILS. Multiply that by concurrency and by worker count and the fleet is holding thousands of records in RAM while the downstream sink accepts a few dozen writes per second. The queue is not the problem; the queue is the shock absorber. The problem is that nothing tells the workers to stop pulling when the sink cannot keep up.
Root Cause
The default worker configuration is optimized for many small, fast, independent tasks — the opposite of catalog writes, which are large and gated by a slow, rate-limited downstream. Celery’s default worker_prefetch_multiplier is 4, so a worker with 8 concurrent slots reserves up to 32 messages at once; raise concurrency and prefetch scales with it. Each reserved message pins its payload in the worker’s memory until the task acknowledges. With acknowledgement happening early (the Celery default acks a task before it runs), a worker greedily reserves a large working set that has no relationship to how fast the ILS or database can absorb it.
This is the classic fast-broker, slow-sink mismatch, and it has no built-in backpressure. Backpressure means the rate of intake is tied to the rate the downstream can absorb — when the sink slows, the source must slow too. A broker deliberately breaks that coupling: it accepts producers at full speed and buffers, which is exactly what you want for smoothing bursts but exactly what removes the natural throttle a synchronous pipeline would have had. The workers, seeing an always-full queue, pull as fast as their prefetch allows and convert a broker backlog (cheap, on disk) into a worker backlog (expensive, in RAM), while hammering an API that is already telling you to slow down through the Retry-After header covered in ILS REST API Polling & Rate Limiting.
Two independent bounds are missing: one on how much work a worker may hold (prefetch and concurrency), and one on how fast the fleet may push to the downstream (a rate limit keyed on the sink). Fixing only the first still lets a large fleet overwhelm a single-writer database; fixing only the second still lets each worker balloon in memory. You need both, plus a shed valve for when the backlog itself grows unbounded.
Solution
Bound the pipeline on three axes. First, make each worker hold exactly one in-flight task by setting worker_prefetch_multiplier=1 with task_acks_late=True, so a task is acknowledged only after it succeeds and a crashed worker’s task is redelivered rather than lost. Second, cap concurrency so total in-flight work across the fleet cannot exceed what the downstream tolerates. Third, meter writes with a rate limit / token bucket keyed on the downstream, and pause intake when the queue depth crosses a high-water mark. On RQ the equivalents are: run a fixed, small worker count (RQ has no prefetch — it pops one job at a time, which is already prefetch-one), wrap the sink call in the same token bucket, and gate enqueue on queue.count before scheduling more work.
from __future__ import annotations
import logging
import time
from contextlib import contextmanager
from typing import Iterator
import redis
from celery import Celery
logger = logging.getLogger("catalog.backpressure")
app = Celery("catalog")
app.conf.update(
# One reserved task per worker process: memory is bounded to the
# in-flight payload, not prefetch_multiplier * concurrency payloads.
worker_prefetch_multiplier=1,
# Acknowledge only after the task succeeds, so a killed worker's task
# is redelivered instead of silently dropped.
task_acks_late=True,
task_reject_on_worker_lost=True,
# Bound how long a single ILS write may block before we reclaim the slot.
task_soft_time_limit=30,
task_time_limit=45,
# Cap concurrency to what the ILS/DB can absorb, not to CPU count.
worker_concurrency=4,
broker_transport_options={"visibility_timeout": 120},
)
# High-water mark: above this backlog we stop admitting new intake tasks.
QUEUE_HIGH_WATER = 20_000
class DownstreamSaturated(Exception):
"""Raised when the token bucket for the downstream sink is exhausted."""
class TokenBucket:
"""Redis-backed token bucket keyed on a downstream identifier.
One bucket per sink (e.g. the ILS write endpoint) bounds the *aggregate*
write rate across every worker in the fleet, which a per-worker limit
cannot do.
"""
def __init__(self, client: redis.Redis, key: str, rate: float, capacity: int) -> None:
self._client = client
self._key = f"bucket:{key}"
self._rate = rate # tokens refilled per second
self._capacity = capacity # max burst
def acquire(self) -> None:
now = time.monotonic()
with self._client.pipeline() as pipe:
pipe.hmget(self._key, "tokens", "ts")
tokens_raw, ts_raw = pipe.execute()[0]
tokens = float(tokens_raw) if tokens_raw else float(self._capacity)
ts = float(ts_raw) if ts_raw else now
tokens = min(self._capacity, tokens + (now - ts) * self._rate)
if tokens < 1.0:
raise DownstreamSaturated(self._key)
self._client.hset(self._key, mapping={"tokens": tokens - 1.0, "ts": now})
@contextmanager
def downstream_slot(bucket: TokenBucket, *, batch_id: str) -> Iterator[None]:
"""Meter one write against the downstream, or countdown-retry the task."""
try:
bucket.acquire()
except DownstreamSaturated:
logger.info(
"downstream_saturated",
extra={"batch_id": batch_id, "action": "retry_later"},
)
raise
yield
@app.task(
bind=True,
max_retries=None,
# Per-task fallback limit; the token bucket is the real throttle.
rate_limit="30/s",
retry_backoff=True,
retry_backoff_max=60,
retry_jitter=True,
)
def apply_catalog_update(self, bib_id: str, marc_blob: str, *, batch_id: str) -> None:
"""Write one bib record to the ILS under aggregate backpressure."""
client = redis.Redis.from_url("redis://localhost:6379/0")
bucket = TokenBucket(client, key="ils.write", rate=25.0, capacity=50)
# Shed intake at the source: if the backlog is already past the high-water
# mark, defer this task so the fleet drains before pulling more.
backlog = client.llen("catalog.updates")
if backlog > QUEUE_HIGH_WATER:
logger.warning(
"backlog_high_water",
extra={"backlog": backlog, "batch_id": batch_id, "action": "pause"},
)
raise self.retry(countdown=15)
try:
with downstream_slot(bucket, batch_id=batch_id):
_write_to_ils(bib_id, marc_blob)
except DownstreamSaturated as exc:
# Bucket empty: hand the slot back and try again shortly.
raise self.retry(exc=exc, countdown=5)
except TimeoutError as exc:
logger.error(
"ils_write_timeout",
extra={"bib_id": bib_id, "batch_id": batch_id},
)
raise self.retry(exc=exc)
The behavioural change is that intake is now coupled to the sink. Before, each worker reserved a large batch and pushed until the ILS collapsed; memory scaled with prefetch and concurrency. After, a worker holds one task, writes only when the shared token bucket grants a slot, and stops admitting new work entirely once the backlog crosses the high-water mark — the pause arrow in the diagram. A task that cannot get a token is not dropped; it is re-queued with a short countdown, so the broker resumes its proper role as the shock absorber and the workers stay flat in memory. Tasks that exhaust retries against a persistently unavailable sink should route to a dead-letter queue, as described in Recovering Dead-Letter Queue Messages in Catalog Ingestion, rather than retrying forever.
Compliance or Privacy Impact
Backpressure is a reliability control, but it touches privacy in two ways worth tracking. First, the payloads buffered under high prefetch are catalog records, not patron data — bibliographic MARC is not PII — so a memory balloon here does not by itself widen the patron-data surface. But the same worker fleet is often reused for patron-facing syncs, and there the buffered payloads do carry personal fields. Keep patron-sync tasks on a separate queue with its own bounded pool so a catalog burst can never pin patron records in the memory of a crashed worker, and so masked-export guarantees from PII Masking in Patron Data Exports are not undermined by a task that dies mid-write and redelivers a half-masked payload.
Second, task_acks_late=True means a task can run more than once — on redelivery after a worker loss, or after a retry. Any write to a downstream that carries personal data must therefore be idempotent, keyed on a stable record identifier, so a redelivered task overwrites rather than duplicates. Log the backpressure decisions (backlog_high_water, downstream_saturated) with a batch_id but never with record contents, so the operational audit trail does not itself become a copy of the data you are trying to protect.
Verification
Confirm the bound holds under a synthetic burst: enqueue far more tasks than the sink can absorb and assert that worker memory stays flat and the downstream is never asked to exceed its rate. A load test with a stubbed sink makes the token bucket’s ceiling observable.
import threading
import time
import redis
def test_token_bucket_caps_aggregate_rate() -> None:
client = redis.Redis.from_url("redis://localhost:6379/15")
client.delete("bucket:test.sink")
bucket = TokenBucket(client, key="test.sink", rate=25.0, capacity=25)
granted = 0
lock = threading.Lock()
def hammer() -> None:
nonlocal granted
for _ in range(200):
try:
bucket.acquire()
with lock:
granted += 1
except DownstreamSaturated:
pass
start = time.monotonic()
threads = [threading.Thread(target=hammer) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
elapsed = time.monotonic() - start
# Across 8 concurrent workers the bucket must never exceed
# capacity + rate * elapsed grants — proving an aggregate ceiling.
ceiling = 25 + 25.0 * elapsed
assert granted <= ceiling, (granted, ceiling)
For a running pipeline, watch three signals together during an overlay load: the broker backlog (redis-cli LLEN catalog.updates) should rise then fall smoothly rather than the workers falling over; worker maxrss should stay flat and proportional to one payload times concurrency, not to prefetch times concurrency; and the ILS 429 rate should drop to near zero because the token bucket now caps the fleet below the API’s limit. If backlog climbs without ever draining, the sink is genuinely slower than intake and you need a lower rate or a bigger dead-letter escape valve — not a bigger worker pool, which only moves the balloon back into memory.
Related
- Async Batch Processing for Catalog Updates — the parent guide to the distributed-ingestion contract these bounds protect.
- Catalog Ingestion & ILS Sync Pipelines — the wider ingestion architecture and its broker patterns.
- Using Celery for Distributed Catalog Ingestion — the baseline worker topology this page tunes.
- Recovering Dead-Letter Queue Messages in Catalog Ingestion — where tasks go when the downstream stays saturated past their retry budget.