Bài 64: Checklist & Anti-patterns Queue — Incident PEL Leak

Mục Tiêu Bài Học

• Phân tích incident PEL leak: chuỗi nguyên nhân, hậu quả, và cách fix.
• Nhận diện 10 anti-pattern phổ biến khi dùng Redis làm queue.
• Áp dụng checklist production-ready trước khi deploy hệ thống queue.
• Chọn đúng cấu trúc dữ liệu (List, Streams, Sorted Set, Pub/Sub, Kafka) cho từng use case.
• Tự kiểm tra kiến thức Module 5 trước khi học Module 6.

Incident: PEL Leak — 10.000 Message Kẹt Một Tuần

Bối cảnh hệ thống

Hệ thống gửi notification (email và push) cho người dùng sau khi họ đặt hàng. Kiến trúc:

• Producer ghi event vào Redis Stream stream:notifications.
• Consumer Group notif-workers với 5 worker pod, mỗi pod xử lý khoảng 100 message/s.
• Mỗi pod có hostname riêng (ví dụ worker-pod-abc123), dùng làm consumer name.
• Code chính: XREADGROUP GROUP notif-workers <hostname> COUNT 10 BLOCK 5000 STREAMS stream:notifications > → process → XACK.

Điểm thiếu: không có lệnh XAUTOCLAIM nào trong codebase.

Diễn biến

1. Hệ thống hoạt động bình thường. Không có alert queue nào.
2. Trong quá trình rolling deploy và K8s pod eviction thường xuyên, các pod restart vài lần mỗi ngày.
3. Mỗi lần pod restart: một số message đang trong vòng xử lý (sau XREADGROUP, chưa XACK) bị bỏ lại trong PEL của consumer name cũ.
4. Pod mới khởi động với hostname mới → consumer name mới → không nhận PEL của consumer cũ.
5. Consumer cũ (worker-pod-abc123) không còn tồn tại → PEL của nó không ai nhận.
6. Sau một tuần: XPENDING trả về hơn 10.000 message pending chia cho nhiều consumer name "ma" không còn tồn tại.

Triệu chứng

User complaint: "Tôi đặt hàng 3 tiếng trước, vẫn chưa nhận email xác nhận." Một số đơn chờ email nhiều giờ, một số không bao giờ nhận vì message bị kẹt vĩnh viễn.

Inspect bằng XPENDING stream:notifications notif-workers:

1) (integer) 10847          # tổng pending
2) "1701000001000-0"        # min ID (oldest)
3) "1701604800000-9"        # max ID (newest)
4) 1) 1) "worker-pod-abc123"
      2) "2134"
   2) 1) "worker-pod-def456"
      2) "1987"
   # ... thêm 8 consumer name khác

Kiểm tra các consumer còn sống:

XINFO CONSUMERS stream:notifications notif-workers

Kết quả: tất cả 10 consumer name trong XPENDING đều có idle time hàng giờ, nhiều consumer không còn pod tương ứng.

Root Cause Phân Tích

Chuỗi nguyên nhân

Ba quyết định thiết kế kết hợp tạo ra sự cố:

1. Consumer name theo pod hostname. Kubernetes đặt tên pod ngẫu nhiên (hoặc theo index trong StatefulSet). Khi pod restart, hostname thay đổi → consumer name mới → PEL của consumer cũ không được kế thừa.

Nếu dùng consumer name cố định theo vai trò (ví dụ worker-0, worker-1...), pod restart cùng consumer name → PEL tự động kế thừa. Đây là cách StatefulSet + consumer name theo ordinal giải quyết vấn đề.

2. Không có recovery loop (XAUTOCLAIM). Code chỉ gọi XREADGROUP với > (nhận message mới), không bao giờ check PEL của consumer khác. Bài 57 đã giải thích: Redis không tự recovery — application phải chủ động gọi XCLAIM hoặc XAUTOCLAIM.

3. Không monitor PEL size. Không có metric nào theo dõi XPENDING count. PEL tăng dần trong 7 ngày mà không có alert nào kích hoạt.

Tại sao đây là lỗi thiết kế, không phải lỗi Redis

Redis Streams được thiết kế để application chịu trách nhiệm recovery. At-least-once delivery đòi hỏi cả hai phía: Redis giữ message trong PEL cho đến khi XACK, và application phải có cơ chế XCLAIM/XAUTOCLAIM để lấy lại message từ consumer dead. Thiếu một trong hai thì at-least-once thành at-most-once với risk mất message.

Fix: Bốn Layer Phục Hồi

Layer 1 — Thêm XAUTOCLAIM trong worker loop

Mỗi vòng lặp worker, sau khi xử lý message mới, chạy thêm một lần claim message stale:

import redis
import time
import os

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

STREAM = "stream:notifications"
GROUP = "notif-workers"
CONSUMER = os.environ.get("CONSUMER_NAME", f"worker-{os.getpid()}")
CLAIM_INTERVAL = 30        # giây
IDLE_THRESHOLD_MS = 60_000  # 60 giây không XACK = stale

last_claim_time = 0

def worker_loop():
    global last_claim_time
    claim_cursor = "0-0"

    while True:
        # Nhận message mới
        results = r.xreadgroup(
            GROUP, CONSUMER,
            {STREAM: ">"},
            count=10, block=5000
        )
        if results:
            for stream_name, messages in results:
                for msg_id, data in messages:
                    process_message(msg_id, data)
                    r.xack(STREAM, GROUP, msg_id)

        # Định kỳ claim message stale từ consumer khác
        now = time.time()
        if now - last_claim_time >= CLAIM_INTERVAL:
            claimed_cursor, claimed_msgs = r.xautoclaim(
                STREAM, GROUP, CONSUMER,
                min_idle_time=IDLE_THRESHOLD_MS,
                start_id=claim_cursor,
                count=50
            )
            for msg_id, data in claimed_msgs:
                if data:  # data=None = message đã bị delete
                    process_message(msg_id, data)
                    r.xack(STREAM, GROUP, msg_id)
            claim_cursor = claimed_cursor
            if claim_cursor == "0-0":  # sweep xong một vòng
                last_claim_time = now

Layer 2 — Detect và xóa consumer "ma"

Chạy định kỳ (cronjob mỗi 5 phút) để phát hiện consumer idle quá lâu và không còn pod tương ứng:

def cleanup_ghost_consumers(max_idle_ms=300_000):
    """Xóa consumer không còn pod tương ứng, idle > 5 phút."""
    consumers = r.xinfo_consumers(STREAM, GROUP)
    active_pods = get_active_pod_names()  # từ K8s API hoặc health check endpoint

    for c in consumers:
        name = c["name"]
        idle = c["idle"]
        pending = c["pending"]

        if idle > max_idle_ms and name not in active_pods:
            if pending > 0:
                # XAUTOCLAIM trước khi delete để không mất message
                r.xautoclaim(STREAM, GROUP, "cleanup-worker",
                             min_idle_time=max_idle_ms, start_id="0-0")
            r.xgroup_delconsumer(STREAM, GROUP, name)

Lưu ý: chỉ xóa consumer sau khi đã claim hết PEL của consumer đó sang consumer khác.

Layer 3 — Stable consumer name

Thay vì dùng hostname ngẫu nhiên, dùng consumer name ổn định:

• Kubernetes StatefulSet: pod tên worker-0, worker-1... → consumer name worker-0, worker-1.
• Deployment thông thường: gán consumer name qua environment variable (CONSUMER_NAME=worker-0) khi deploy.

Với stable name, pod restart → cùng consumer name → PEL cũ tự động được kế thừa bởi pod mới.

Layer 4 — Monitor và alert

# Script lấy PEL size per consumer, export ra metrics (Prometheus/Datadog)
def get_pel_metrics():
    info = r.xpending(STREAM, GROUP)
    total_pending = info["pending"]

    # Per-consumer breakdown
    consumers_pending = {}
    if info["consumers"]:
        for consumer_name, count in info["consumers"].items():
            consumers_pending[consumer_name] = int(count)

    return total_pending, consumers_pending

# Alert khi PEL tổng > 1000 và tăng liên tiếp trong 10 phút
# Alert khi có consumer pending > 500 và idle > 5 phút

Top 10 Anti-patterns Queue

1. PEL Leak — Thiếu XAUTOCLAIM

Biểu hiện: Dùng Redis Streams + Consumer Group nhưng không có recovery loop. Worker restart thường xuyên (rolling deploy, OOM, eviction).

Hậu quả: Message tích lũy trong PEL của consumer dead, không được xử lý.

Fix: Bài 57 — XAUTOCLAIM trong worker loop, stable consumer name, monitor PEL size.

2. At-most-once Cho Task Không Được Phép Mất

Biểu hiện: Dùng RPOP để lấy message → xử lý → crash giữa chừng. Message đã bị pop khỏi List, không có PEL, không có cách recover.

Hậu quả: Message mất vĩnh viễn khi worker crash sau pop nhưng trước khi xử lý xong.

Fix: Dùng LMOVE kết hợp processing queue (List làm reliable queue — bài 54), hoặc chuyển sang Redis Streams (bài 55).

3. Consumer Không Idempotent + Có Retry

Biểu hiện: Message được delivery nhiều lần (XAUTOCLAIM claim lại, hoặc crash sau XREADGROUP trước XACK). Consumer xử lý không idempotent — mỗi lần xử lý tạo ra một side-effect mới.

Hậu quả: User nhận email trùng, đơn hàng bị tạo hai lần, tiền bị trừ hai lần.

Fix: Bài 61 — idem key với TTL phù hợp, DB unique constraint cho critical write.

4. Immediate Retry Không Có Backoff

Biểu hiện: Message fail → retry ngay lập tức → fail → retry ngay → vòng lặp không dừng, spam downstream service đang gặp sự cố.

Hậu quả: Downstream không có thời gian recover, tình trạng tệ hơn (cascade failure).

Fix: Bài 59 — exponential backoff với jitter. Ví dụ: lần 1 sau 1s, lần 2 sau 2s, lần 3 sau 4s, có jitter ±20%.

5. Retry Vô Hạn

Biểu hiện: Không có giới hạn retry count. Message fail liên tục (do bug trong code xử lý, hoặc dữ liệu corrupt) → retry mãi mãi, chiếm tài nguyên worker.

Hậu quả: Worker bị chiếm bởi poison message, throughput giảm, message mới bị chậm.

Fix: Bài 59, 60 — max retry limit (ví dụ 5 lần), sau đó đẩy sang DLQ để debug thủ công.

6. Stream Không Cap MAXLEN

Biểu hiện: XADD stream:orders * ... không kèm MAXLEN. Consumer chậm hơn producer → stream grow vô hạn.

Hậu quả: Redis OOM, hoặc eviction xóa key stream toàn bộ (nếu maxmemory-policy không phải noeviction).

Fix: XADD stream:orders MAXLEN ~ 100000 * .... Ký hiệu ~ cho phép Redis trim approximate để tránh overhead. Chọn MAXLEN dựa trên throughput và retention time cần thiết.

7. Không Có DLQ

Biểu hiện: Message sau khi vượt max retry bị XACK và bỏ (drop), hoặc không có DLQ stream để debug.

Hậu quả: Silent data loss — hệ thống trông bình thường nhưng một số message bị mất không trace được.

Fix: Bài 60 — DLQ stream riêng (stream:notifications:dlq), ghi đầy đủ metadata (error, delivery count, original payload), có alert khi DLQ có message mới.

8. Stream Không Persist (AOF Off)

Biểu hiện: Redis chạy với appendonly no (chỉ RDB snapshot). Redis restart giữa hai snapshot → mất toàn bộ message chưa xử lý trong stream.

Hậu quả: Mất message có thể kéo dài từ vài giây đến vài phút tùy RDB interval.

Fix: Bật appendonly yes với appendfsync everysec cho balance giữa durability và performance. Với message cực kỳ critical, cân nhắc Kafka thay Redis.

9. Worker Không Graceful Shutdown

Biểu hiện: Worker bị SIGKILL hoặc dừng đột ngột khi đang giữa xử lý. Message đã XREADGROUP nhưng chưa XACK → vào PEL. Nếu không có XAUTOCLAIM thì kẹt (anti-pattern 1).

Hậu quả: Mỗi deploy thêm một lượng message vào PEL. Với nhiều deploy mỗi ngày, PEL tích lũy nhanh.

Fix: Handle SIGTERM — dừng nhận message mới, finish message đang xử lý, XACK, rồi thoát. Kubernetes gracefulTerminationPeriod đủ để worker finish batch hiện tại.

10. Pub/Sub Làm Queue

Biểu hiện: Dùng PUBLISH / SUBSCRIBE để gửi task (ví dụ: publish event "send-email").

Hậu quả: Pub/Sub là fire-and-forget — không có persistence, không có acknowledgment. Subscriber offline → message mất. Pod restart → message mất. Không có retry, không có DLQ.

Fix: Dùng Redis Streams cho task cần reliable delivery. Pub/Sub phù hợp cho real-time broadcast khi mất message là chấp nhận được (Module 6 — bài 65 sẽ phân tích rõ use case đúng của Pub/Sub).

Checklist Production-ready Queue

Checklist này áp dụng trước khi đưa bất kỳ hệ thống queue Redis vào production. Mỗi mục là một câu hỏi cụ thể cần trả lời "có" hoặc "không áp dụng với lý do rõ ràng".

Reliability

• Stream + Consumer Group (không dùng List RPOP cho task không được mất)?
• XACK ngay sau khi xử lý thành công?
• XAUTOCLAIM chạy định kỳ trong worker loop (mỗi 10–30 giây)?
• Stream có MAXLEN cap để tránh grow vô hạn?
• AOF bật (appendonly yes)?
• Worker có graceful shutdown (handle SIGTERM, finish batch trước khi thoát)?

Retry

• Error được phân loại: retryable (network timeout, temporary fail) vs non-retryable (validation error, bad data)?
• Retry dùng exponential backoff với jitter?
• Có max retry limit (ví dụ 3–5 lần)?
• Message vượt max retry đi vào DLQ (không drop silently)?

Idempotency

• Consumer xử lý idempotent — chạy N lần cùng message cho ra kết quả giống chạy 1 lần?
• Idem key được lưu với TTL phù hợp (lớn hơn max retry window)?
• DB write critical có unique constraint hoặc upsert?

Scaling

• Consumer name unique và ổn định per process (không dùng random hostname)?
• Số partition stream và số worker phù hợp với throughput target?
• Có cơ chế autoscale consumer dựa trên backlog metric (stream length, PEL size)?

Monitoring

• PEL size per consumer được export thành metric?
• Stream length (backlog) được theo dõi?
• DLQ ingestion rate có alert khi spike?
• Throughput per worker được đo?
• Alert khi PEL tăng liên tục mà không giảm (dấu hiệu XAUTOCLAIM không hoạt động)?
• Alert khi DLQ nhận message (mọi message vào DLQ đều cần được điều tra)?

Decision Tree — Chọn Cấu Trúc Queue

Câu hỏi đầu tiên: yêu cầu delivery semantics là gì?

Nếu mất message là chấp nhận được

• Analytics event, log aggregation, metrics sampling: List BRPOP đơn giản. Nếu consumer chậm, dùng MAXLEN để cap.
• Real-time broadcast đến nhiều subscriber: Pub/Sub (Module 6). Fire-and-forget, subscriber offline → mất message.

Nếu cần reliable delivery

• Task đơn giản, không cần consumer group, throughput thấp: List + LMOVE làm processing queue (bài 54). Đơn giản hơn Streams nhưng không có PEL per consumer, recovery thủ công hơn.
• Task quan trọng, multi-worker, cần at-least-once, cần PEL: Redis Streams + Consumer Group (bài 55, 56, 57). Nên dùng cho phần lớn production use case với Redis.
• Task cần chạy sau một khoảng thời gian (delayed/scheduled): Sorted Set với score là timestamp (bài 58). Kết hợp với Streams để reliable delivery sau khi dequeue.
• Priority queue: Multiple streams với weighted polling, hoặc Sorted Set score theo priority (bài 61).

Nếu Redis không đủ

• Throughput cực lớn (hàng triệu message/s), retention dài, replay: Kafka. Redis Streams có thể handle hàng trăm nghìn message/s, nhưng Kafka có partition replication, consumer offset management, và replay từ offset bất kỳ mà Redis không có.
• Routing phức tạp, multiple exchange type, dead-lettering tích hợp sẵn: RabbitMQ. Bài 63 đã so sánh chi tiết.
• Exactly-once semantics với transaction: Kafka (transactional producer + idempotent consumer). Redis không có exactly-once — chỉ at-least-once.

Tóm lại về phạm vi phù hợp của Redis queue: throughput vừa đến cao (đến vài trăm nghìn msg/s), latency thấp (sub-millisecond), retention ngắn (giờ đến ngày), team quen Redis hơn Kafka/RabbitMQ. Khi một trong các điều kiện này không thỏa mãn, xem xét message broker chuyên dụng.

Module 5 — Tổng Kết Khái Niệm

Module 5 gồm 12 bài (53–64) xây dựng từ câu hỏi "vì sao cần queue" đến hệ thống queue production-ready hoàn chỉnh.

Self-assessment Trước Module 6

Kiểm tra lại kiến thức trước khi chuyển sang Module 6. Mỗi câu hỏi nên trả lời được không cần tra cứu.

Delivery semantics

• At-most-once, at-least-once, và exactly-once khác nhau như thế nào? Cấu trúc Redis nào cho từng loại?
• Tại sao Redis Streams + Consumer Group chỉ đảm bảo at-least-once chứ không phải exactly-once?

Stream vs List

• Cho use case nào thì List đủ, cho use case nào thì cần Streams?
• Khi nào cần LMOVE thay RPOP?

Recovery

• Giải thích cơ chế PEL. Tại sao message không tự thoát khỏi PEL?
• XAUTOCLAIM hoạt động như thế nào? min-idle-time là gì?
• Stable consumer name quan trọng như thế nào trong rolling deploy?

Retry & DLQ

• Thiết kế retry strategy cho một task gọi external API: retry những lỗi nào, backoff như thế nào, max bao nhiêu lần?
• Khi nào message nên vào DLQ thay vì tiếp tục retry?

Giới hạn Redis

• Kể 3 tình huống nên dùng Kafka thay Redis Streams.
• Pub/Sub phù hợp cho use case nào? Không phù hợp cho use case nào?

Câu hỏi thiết kế

Hệ thống order processing: sau khi user đặt hàng, cần (1) gửi email xác nhận, (2) cập nhật inventory, (3) tạo invoice PDF. Ba task này có thể fail độc lập, cần retry, email phải exactly-once (không gửi trùng). Thiết kế queue như thế nào? Dùng một hay nhiều stream? Xử lý idempotency cho email như thế nào?

Bài Tiếp Theo

Module 6 — Real-time Systems bắt đầu từ bài 65: Pub/Sub Cơ Bản — Fire-and-forget.

Module 6 khác Module 5 ở điểm cốt lõi: real-time broadcast chấp nhận mất message khi subscriber offline. Pub/Sub không có PEL, không có acknowledgment, không có replay. Đây là trade-off có chủ ý — đổi reliability lấy latency cực thấp và fan-out không giới hạn. Bài 65 sẽ phân tích rõ use case đúng và sai của Pub/Sub để tránh nhầm lẫn với Streams.

Tham khảo

• Redis Docs — Streams Introduction
• Redis Docs — XAUTOCLAIM
• Redis Docs — XPENDING
• Redis Docs — XINFO CONSUMERS
• Redis Docs — XGROUP DELCONSUMER
• Redis Docs — XADD (MAXLEN option)
• Redis Blog — Real-time System Patterns