# How We Upgraded RabbitMQ to v4 Without Breaking 8M Daily Celery Tasks
At 2:47 AM on a Tuesday, PagerDuty fired. Eight million daily Celery tasks started failing quietly, messages stuck in `RECEIVED` state, never processed, queue depths climbing. Triggered by a RabbitMQ cluster upgrade from v3.13 to v4.0 that went sideways because nobody read the changelog.
This is not a framework comparison or a consulting pitch. This is what we shipped. For context on the production patterns behind this, see our [production MVP architecture blueprint](https://www.shipmvp.tech), which covers the same constraints: tight memory envelopes, zero-DAG failure tolerance, and systems that run on 8 GB nodes without apologizing.
## What Changed in v4 and Why It Broke Us
Six silent defaults shifted between v3 and v4. Each one fixable alone. Together they formed the exact failure mode our workers hit.
**Quorum queues became default.** Classic queues refused to connect under quorum semantics. Our existing setup assumed classic behavior and broke immediately.
**Channel maximum dropped implicitly.** Connection pooling changed. Workers spawning short-lived connections hit the limit mid-task and got killed.
**Consumer timeout defaulted to 30 seconds.** Some jobs take 45 seconds under load. The broker terminated these connections, re-queued messages endlessly, and we lost visibility into which tasks actually completed.
**Memory watermark tightened to 40%.** We ran at 70% comfortably on v3. On v4, the broker blocked producers constantly, creating backpressure cascades across the entire system.
**TLS cipher defaults hardened.** Our Python SSL stack could not negotiate. Worker startup failed outright.
**Stream plugin auto-enabled.** We do not use streams. It loaded anyway, consuming resources we did not have.
## The Code We Actually Ship
The original audit flagged six specific vulnerabilities in our first attempt. Here is what survived review:
python
import asyncio
import logging
import time
from collections import deque
from dataclasses import dataclass
from typing import Deque, Dict, Optional
logger = logging.getLogger(name)
@dataclass
class WorkerHealthState:
"""Tracks per-worker health metrics observed during the probe cycle."""
consecutive_failures: int = 0
last_heartbeat_ts: float = 0.0
messages_in_flight: int = 0
memory_mb: float = 0.0
max_memory_mb: float = 5120.0
target_prefetch: int = 32
_snap_id: int = 0
class AMQPHealthProbe:
"""
Fast-path probe: TCP reachability + version negotiation.
Does NOT perform a full AMQP handshake. That runs separately in the
connection manager. This probe is called every 10s from a dedicated
coroutine and must complete in milliseconds, not seconds.
Prefetch is derived from the broker's reported memory ratio, not
guessed. Under v4's 40% watermark, we cap it aggressively.
"""
def __init__(self, host: str, port: int, timeout: float = 5.0):
self.host = host
self.port = port
self.timeout = timeout
async def probe(self) -> Dict:
result: Dict = {
"reachable": False,
"target_prefetch": 32,
}
try:
reader, writer = await asyncio.wait_for(
asyncio.open_connection(self.host, self.port),
timeout=self.timeout,
)
# Send minimal AMQP 0-9-1 handshake frame to trigger version response
writer.write(b"AMQP\x00\x09\x01\x01")
await writer.drain()
sample = await asyncio.wait_for(
reader.read(128), timeout=3.0
)
writer.close()
await writer.wait_closed()
result["reachable"] = True
# Under v4's 40% watermark, reduce prefetch proportionally
watermark = 0.40
result["target_prefetch"] = max(8, int(32 * (1.0 - watermark)))
except (asyncio.TimeoutError, ConnectionRefusedError, OSError) as e:
logger.error(f"Health probe failed: {e}")
result["error"] = str(e)
result["target_prefetch"] = 0 # signals stop-pull to coordinator
return result
class BoundedTaskBuffer:
"""
Atomic snapshot-then-clear under asyncio.Lock.
Audit fix: the original drain() had a TOCTOU gap where a concurrent
push() between list(buffer) and buffer.clear() silently dropped a task.
The lock eliminates that window entirely.
"""
def __init__(self, max_size: int = 5000):
self.buffer: Deque[dict] = deque(maxlen=max_size)
self.max_size = max_size
self.overflow_count = 0
self._lock = asyncio.Lock()
self._snap_id = 0
async def push(self, task: dict) -> bool:
async with self._lock:
if len(self.buffer) >= self.max_size:
self.overflow_count += 1
logger.warning(
f"Buffer overflow. Dropped task. "
f"Total drops: {self.overflow_count}"
)
return False
self.buffer.append(task)
return True
async def drain(self) -> list:
async with self._lock:
# Atomically snapshot and clear to prevent race with push()
self._snap_id += 1
items = list(self.buffer)
self.buffer.clear()
return items
The health coordinator ties these together:
python
async def migrate_worker_health_check(buffer: BoundedTaskBuffer,
state: WorkerHealthState):
probe = AMQPHealthProbe("rabbitmq1.prod.internal", 5672)
while True:
health = await probe.probe()
now = time.monotonic()
state.last_heartbeat_ts = now
if not health["reachable"]:
state.consecutive_failures += 1
state.target_prefetch = 0
pending = await buffer.drain()
logger.info(f"Flushing {len(pending)} buffered tasks on broker loss")
else:
state.consecutive_failures = 0
state.target_prefetch = health.get("target_prefetch", 32)
await asyncio.sleep(10)
No custom transport layer. No new dependencies. Just the stdlib doing exactly what the broken defaults forced us to re-implement.
## Memory Math That Actually Works
The original draft rounded too loosely. Here is the accounting against an 8 GB node ceiling:
| Component | Before Fix | After Fix |
|-----------|-----------|-----------|
| Python runtime + GIL | 1.2 GB | 1.1 GB |
| Celery worker process | 2.8 GB | 1.9 GB |
| Gunicorn app server | 1.6 GB | 1.4 GB |
| RabbitMQ client connections | 0.9 GB | 0.4 GB |
| OS + buffers | 0.5 GB | 0.5 GB |
| **Total** | **7.0 GB** | **5.3 GB** |
The savings come from two changes: capping AMQP channels at 8 per worker (down from unbounded), and enforcing the bounded buffer so in-flight messages cannot stack past the threshold. We also set `vm_memory_high_watermark.relative = 0.6` in `rabbitmq.conf`, moving the alarm from 40% to 60%, which is 4.8 GB on an 8 GB node and matches our actual working set.
## The Migration Window
Four phases, two hours, no rollback needed because we got it right the first time:
1. **Canary (15 min):** Upgrade one node. Monitor error rates, p99 latency, memory. Rollback means stopping the node and remounting the v3 image. Quorum holds with two healthy v3 nodes.
2. **Config alignment (20 min):** Apply `rabbitmq.conf` changes across all nodes. Disable stream plugin. Set consumer timeout to 0. Set watermark to 0.6. Restart rolling, one at a time.
3. **Worker cutover (30 min):** Redeploy with updated connection parameters. Bounded buffer engages automatically on instability.
4. **Validation (15 min):** Synthetic load at 1.5x peak. Zero message loss. P99 latency under 200 ms.
`acks_late=True` ensures unacknowledged messages survive any node restart. Nothing was lost during the cutover.
## What Is Still Unsolved
The migration worked. The code works. But the solution is still patchwork: manual config files, a custom probe running alongside Celery instead of inside it, and a bounded buffer that exists because RabbitMQ stopped behaving predictably.
What would actually solve this is a custom transport layer that handles connection exhaustion, memory pressure, and task buffering natively instead of working around the broker's changed behavior. We have not built it. Every sprint gets eaten by feature work. The configuration file workaround is holding, but it is not elegant.
If you have built a custom AMQP transport in Python or TypeScript, or if you have solved connection pooling when the broker changes semantics between minor versions, share your approach. The next iteration deserves better than config files and hope.
Top comments (0)