HARRY-23 — GĐ9 — Queues, Jobs, Workers & Cronjob: xử lý nền đúng cách

GĐ9 — Queues, Jobs, Workers & Cronjob: xử lý nền đúng cách

GĐ8 đã giới thiệu BullMQ và outbox ở mức "chạy được". Giai đoạn này đi sâu vào những thứ làm sập production lúc 2 giờ sáng: job chạy hai lần, cron chạy trên cả 5 instance, worker bị kill giữa chừng, queue tắc nghẽn vì một tenant, và job "đã xong" nhưng dữ liệu chỉ hoàn thành một nửa.
Đây là giai đoạn phân biệt rõ nhất giữa "biết dùng thư viện" và "hiểu hệ thống bất đồng bộ". Nó cũng là nền tảng bắt buộc trước khi vào GĐ16 — Distributed Systems.

1. Vì sao cần xử lý nền

Vấn đề. Request HTTP có ngân sách thời gian: người dùng chờ, load balancer timeout (thường 30–60s), trình duyệt bỏ cuộc. Nhưng nhiều việc không vừa ngân sách đó: gửi 10.000 email, resize video, gọi LLM 40 giây, sinh báo cáo PDF.

Giải pháp. Tách "nhận yêu cầu" khỏi "làm việc":

Client ──POST──► API ──đẩy job──► Queue ──lấy job──► Worker ──► kết quả
         ◄─202──┘  (trả về ngay)                    (process riêng)

Ba lợi ích, nói cho rõ:

Độ trễ — người dùng nhận 202 Accepted trong 20ms thay vì chờ 40 giây.
Chống chịu — worker chết thì job vẫn nằm trong queue, chạy lại được. Nếu làm inline, request chết là mất việc.
Điều tiết tải — spike 10.000 request không giết DB; queue giữ lại và worker rút với tốc độ bạn kiểm soát.

Cái giá. Bạn đổi một hệ thống đồng bộ dễ hiểu lấy một hệ thống bất đồng bộ khó debug: job chạy hai lần, chạy sai thứ tự, chạy 3 tiếng sau, hoặc không chạy mà không ai biết. Phần còn lại của giai đoạn này là về việc trả cái giá đó cho đúng.

Việc nào nên đẩy nền:

Nên	Không nên
Gửi email/SMS/push	Xác thực, phân quyền
Xử lý ảnh/video, sinh PDF	Bất cứ thứ gì người dùng cần thấy kết quả ngay
Gọi API bên thứ ba chậm	Ghi dữ liệu mà bước sau phụ thuộc trực tiếp
Tính toán tổng hợp, báo cáo	Việc nhỏ hơn ~50ms (chi phí queue lớn hơn lợi ích)
Đồng bộ sang search index

2. Chọn công nghệ hàng đợi

Công nghệ	Mô hình	Đảm bảo	Dùng khi
BullMQ (Redis)	Job queue	At-least-once	Node monolith. Mặc định của lộ trình này
pg-boss (Postgres)	Job queue trên SQL	At-least-once	Đã có Postgres, không muốn thêm Redis
Bảng SQL tự viết	`FOR UPDATE SKIP LOCKED`	At-least-once	Tải nhỏ, muốn hiểu cơ chế, muốn job trong cùng transaction với dữ liệu
RabbitMQ	Message broker (AMQP)	At-least-once, routing mạnh	Nhiều service, cần routing/fanout phức tạp
Kafka	Log phân tán	At-least-once, giữ thứ tự trong partition, replay được	Event streaming, throughput rất lớn, nhiều consumer group
SQS / Cloud Tasks	Managed	At-least-once (FIFO queue: exactly-once trong 5 phút)	Không muốn vận hành hạ tầng

Khác biệt quan trọng nhất: queue vs log.

Queue (BullMQ, SQS, Rabbit): job được tiêu thụ rồi biến mất. Một job → một worker xử lý.
Log (Kafka): sự kiện được giữ lại, nhiều consumer group đọc độc lập, mỗi group có offset riêng, replay được từ đầu.

Nếu bạn cần "phát lại 3 ngày sự kiện vì service mới vừa lên", bạn cần Kafka chứ không phải BullMQ. Nếu bạn cần "gửi email này một lần", queue đủ và đơn giản hơn nhiều.

Pitfall #1 — chọn Kafka cho DA3. Kafka giải quyết vấn đề của hệ thống có nhiều đội và throughput rất lớn. Với một monolith, nó là chi phí vận hành lớn không đổi lấy gì. Chọn BullMQ, và biết vì sao bạn không chọn Kafka — đó mới là câu trả lời phỏng vấn tốt.

3. Đảm bảo giao nhận: at-most-once / at-least-once / exactly-once

Đây là khái niệm quan trọng nhất của cả giai đoạn.

Đảm bảo	Nghĩa	Đánh đổi
At-most-once	Chạy 0 hoặc 1 lần	Có thể mất việc. Ack trước khi làm
At-least-once	Chạy 1 hoặc nhiều lần	Có thể lặp việc. Ack sau khi làm
Exactly-once	Đúng 1 lần	Không tồn tại ở tầng giao nhận qua mạng

Vì sao exactly-once là không thể. Worker xử lý xong, gửi ACK, rồi chết trước khi ACK tới nơi. Queue không phân biệt được "chết trước khi làm" với "chết sau khi làm, trước khi báo". Nó buộc phải chọn: giao lại (→ lặp) hoặc bỏ qua (→ mất). Không có lựa chọn thứ ba.

Kết luận thực hành: mọi hệ thống thật đều là at-least-once, và bạn đạt hiệu quả exactly-once bằng cách làm handler idempotent. Đây là câu trả lời phỏng vấn chuẩn, và cũng là điều bạn phải thực sự làm trong code.

Ba cách làm handler idempotent:

// A. Khoá idempotency ở tầng DB — chắc chắn nhất
async function handleChargeJob(job: Job<ChargePayload>) {
  const key = `charge:${job.data.orderId}`
  try {
    await db.processedJob.create({ data: { key } })   // UNIQUE constraint
  } catch (e) {
    if (isUniqueViolation(e)) return                  // đã làm rồi → thoát êm
    throw e
  }
  await stripe.charge(job.data)
}

// B. Thao tác tự nhiên idempotent — tốt nhất khi làm được
await db.user.update({ where: { id }, data: { status: 'ACTIVE' } })   // gán, không phải $inc
await s3.putObject({ Key: deterministicKey, Body: data })             // ghi đè cùng key

// C. Đẩy idempotency sang bên thứ ba
await stripe.paymentIntents.create(params, { idempotencyKey: `order-${orderId}` })

Pitfall #2 — dùng jobId làm khoá idempotency. BullMQ tự dọn job đã hoàn thành (removeOnComplete), nên jobId có thể được cấp lại và "đã xử lý chưa" không còn tra được. Khoá idempotency phải sống trong DB của bạn, với vòng đời bạn kiểm soát.

Pitfall #3 — $inc trong job. UPDATE counters SET n = n + 1 chạy hai lần cho kết quả sai. Nếu bắt buộc phải cộng dồn, ghi sự kiện có id duy nhất rồi tổng hợp, đừng cộng trực tiếp.

4. Transactional Outbox — bài toán hai-lần-ghi

Vấn đề. Bạn cần làm hai việc nguyên tử ở hai hệ thống khác nhau:

await db.$transaction(async (tx) => {
  await tx.order.create({ data: order })
  await queue.add('send-confirmation', { orderId })   // ← SAI, cả hai chiều
})

Sai theo chiều thứ nhất: queue.add() thành công, rồi transaction rollback → job gửi email cho một đơn hàng không tồn tại. Sai theo chiều thứ hai: transaction commit, rồi Redis chết trước khi queue.add() xong → đơn hàng tồn tại nhưng không ai được báo, vĩnh viễn.

Giải pháp: outbox. Ghi "ý định gửi" vào cùng database, cùng transaction. Một tiến trình riêng đọc bảng đó và đẩy sang queue.

sql

CREATE TABLE outbox (
  id           BIGSERIAL PRIMARY KEY,
  topic        TEXT        NOT NULL,
  payload      JSONB       NOT NULL,
  created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
  published_at TIMESTAMPTZ,
  attempts     INT         NOT NULL DEFAULT 0
);
CREATE INDEX idx_outbox_pending ON outbox (id) WHERE published_at IS NULL;

// 1. Ghi cùng transaction — nguyên tử thật sự vì cùng một DB
await db.$transaction(async (tx) => {
  const order = await tx.order.create({ data: input })
  await tx.outbox.create({
    data: { topic: 'order.created', payload: { orderId: order.id } },
  })
})

// 2. Relay chạy riêng — SKIP LOCKED cho phép nhiều instance chạy song song an toàn
async function relayOnce() {
  await db.$transaction(async (tx) => {
    const rows = await tx.$queryRaw<OutboxRow[]>`
      SELECT * FROM outbox
      WHERE published_at IS NULL
      ORDER BY id
      LIMIT 100
      FOR UPDATE SKIP LOCKED`
    for (const row of rows) {
      await queue.add(row.topic, row.payload, {
        jobId: `outbox-${row.id}`,          // chống lặp ở tầng queue
        removeOnComplete: 1000,
      })
      await tx.$executeRaw`UPDATE outbox SET published_at = now() WHERE id = ${row.id}`
    }
  })
}

FOR UPDATE SKIP LOCKED giải quyết gì. Nó cho phép relay #2 bỏ qua các hàng relay #1 đang giữ, thay vì xếp hàng chờ. Đây là nền tảng của mọi queue xây trên SQL. Không có nó, chạy nhiều relay là vô nghĩa.

Outbox cho bạn at-least-once, không phải exactly-once — relay có thể chết sau queue.add trước UPDATE. Vì vậy handler vẫn phải idempotent (mục 3). Hai cơ chế này bổ sung nhau, không thay thế nhau.

Khi nào dùng outbox: khi mất sự kiện gây hậu quả thật (thanh toán, đơn hàng, email giao dịch, đồng bộ trạng thái). Khi nào không cần: analytics, log không quan trọng, việc có thể tính lại được.

Dọn dẹp. Bảng outbox mọc mãi. Xoá hàng published_at < now() - interval '7 days' bằng cron, hoặc partition theo ngày (→ GĐ13).

5. Thiết kế job: payload, retry, timeout

Payload phải nhỏ và ổn định.

// TỐT — id + phiên bản
await queue.add('resize-image', { assetId: '01H...', v: 1 })

// XẤU — nhét cả object; dữ liệu đã cũ khi worker chạy, và có thể chứa PII
await queue.add('resize-image', { asset: { ...30 fields... }, user: { email, phone } })

Ba lý do: (1) payload là ảnh chụp lúc enqueue, có thể lỗi thời khi worker chạy 3 phút sau; (2) payload nằm trong Redis, PII trong đó là rủi ro tuân thủ; (3) payload lớn tốn RAM Redis và làm chậm mọi thứ.

Thêm v (version) vào payload ngay từ đầu. Khi bạn đổi hình dạng payload, các job cũ vẫn còn trong queue với hình dạng cũ. Có v thì handler xử lý được cả hai; không có thì bạn phải xả queue hoặc chấp nhận job lỗi.

Retry với exponential backoff + jitter:

await queue.add('send-email', payload, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1000 },  // 1s, 2s, 4s, 8s, 16s
  removeOnComplete: { age: 3600, count: 1000 },
  removeOnFail:     { age: 86400 },
})

Vì sao cần jitter. Nếu 5.000 job cùng fail vì một API sập rồi cùng retry sau đúng 1 giây, bạn tạo ra một đợt tấn công vào chính API vừa hồi phục — gọi là thundering herd. Jitter (ngẫu nhiên hoá độ trễ) làm chúng dàn ra. BullMQ hỗ trợ backoff strategy tuỳ biến; hoặc dùng delay: base * 2**n * (0.5 + Math.random()).

Phân biệt lỗi tạm thời và lỗi vĩnh viễn. Đây là thứ hầu hết code bỏ qua:

try {
  await sendEmail(job.data)
} catch (e) {
  if (isPermanent(e)) {            // 400 Bad Request, email không hợp lệ, record bị xoá
    await markFailedPermanently(job.data)
    return                          // KHÔNG throw → không retry vô ích
  }
  throw e                           // 429, 5xx, timeout mạng → để BullMQ retry
}

Retry một địa chỉ email sai cú pháp 5 lần là lãng phí và làm nhiễu cảnh báo.

Timeout. Mọi job phải có timeout, và nó phải ngắn hơn lockDuration, nếu không queue sẽ tưởng worker chết và giao job cho worker khác trong khi worker đầu vẫn đang chạy → hai worker cùng làm một việc.

new Worker('emails', handler, {
  connection,
  concurrency: 10,
  lockDuration: 60_000,       // queue giữ job 60s
  stalledInterval: 30_000,
})
// handler tự đặt timeout < 60s
const res = await fetch(url, { signal: AbortSignal.timeout(30_000) })

Pitfall #4 — job dài hơn lockDuration. Đây là nguyên nhân số một của "job của tôi chạy hai lần mà tôi không hiểu vì sao". Job dài phải gọi job.extendLock()/updateProgress() định kỳ, hoặc chia nhỏ thành nhiều job.

6. Dead Letter Queue và quan sát

DLQ là nơi job hết số lần retry đi đến. Nguyên tắc: job không bao giờ được biến mất im lặng.

worker.on('failed', async (job, err) => {
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await deadLetter.add('failed-job', {
      queue: job.queueName, name: job.name, data: job.data,
      error: err.message, stack: err.stack, failedAt: new Date().toISOString(),
    })
    logger.error({ jobId: job.id, queue: job.queueName, err: err.message }, 'job dead-lettered')
    metrics.increment('job.dead_letter', { queue: job.queueName })
  }
})

Bốn chỉ số phải theo dõi — và ngưỡng cảnh báo:

Chỉ số	Nghĩa	Cảnh báo khi
Queue depth (waiting)	Số job đang chờ	Tăng đơn điệu trong 15 phút → worker không theo kịp
Job age (oldest waiting)	Job cũ nhất chờ bao lâu	Vượt SLA nghiệp vụ (vd: email > 5 phút)
Failure rate	Tỉ lệ fail / tổng	> 1% hoặc tăng đột biến
DLQ size	Số job chết	> 0 là phải có người xem

Queue depth là chỉ số quan trọng nhất và cũng hay bị bỏ quên nhất. Nó là tín hiệu sớm: throughput worker < throughput enqueue. Nếu chỉ nhìn "job có fail không", bạn sẽ phát hiện vấn đề khi email đã trễ 6 tiếng.

Bảng điều khiển. bull-board cho BullMQ — nhưng phải đặt sau xác thực. Nó cho phép xem payload (có thể có PII) và chạy lại job.

Chạy lại từ DLQ. Phải là hành động có chủ đích của con người, không tự động. Trước khi chạy lại: đã sửa nguyên nhân chưa? Handler có idempotent không? Dữ liệu còn hợp lệ không (job 3 ngày tuổi có thể tham chiếu record đã xoá)?

7. Worker: vòng đời và graceful shutdown

Worker phải là process riêng, không phải cùng process với API. Lý do:

Job nặng CPU chặn event loop → API treo (→ GĐ2).
Scale độc lập: 2 API + 8 worker, hoặc ngược lại.
Deploy độc lập; worker crash không làm sập API.

// worker.ts — entrypoint riêng, Dockerfile chung, command khác
const worker = new Worker('emails', handler, { connection, concurrency: 10 })

let shuttingDown = false
async function shutdown(signal: string) {
  if (shuttingDown) return
  shuttingDown = true
  logger.info({ signal }, 'shutting down worker')
  await worker.close()          // ngừng nhận job mới, CHỜ job đang chạy xong
  await db.$disconnect()
  await connection.quit()
  process.exit(0)
}
process.on('SIGTERM', () => shutdown('SIGTERM'))
process.on('SIGINT',  () => shutdown('SIGINT'))

Vì sao SIGTERM quan trọng. Khi bạn deploy, orchestrator (Docker, ECS, K8s) gửi SIGTERM rồi chờ một khoảng grace period trước khi SIGKILL. Nếu worker không xử lý SIGTERM, job đang chạy bị cắt giữa chừng → dữ liệu dở dang.

Ba con số phải khớp nhau:

thời gian job dài nhất  <  grace period của orchestrator  <  lockDuration

Nếu grace period (K8s mặc định 30s) ngắn hơn job dài nhất, mọi lần deploy đều giết job. Chỉnh terminationGracePeriodSeconds → GĐ15.

concurrency là bao nhiêu? Không có con số ma thuật. Bắt đầu ở 5–10, rồi đo:

Job I/O-bound (gọi API, chờ DB): concurrency cao được, giới hạn thật là connection pool DB.
Job CPU-bound: concurrency > số core là vô ích, còn làm chậm đi.

Pitfall #5 — concurrency vượt connection pool. 20 worker × concurrency 10 = 200 kết nối cùng lúc tới Postgres. Pool 10 → 190 job xếp hàng chờ connection và timeout. Tính tổng: số worker × concurrency ≤ pool size (hoặc dùng PgBouncer → GĐ18).

8. Cronjob và scheduled job

Đây là phần yếu nhất trong hầu hết codebase. Mọi người viết setInterval(cleanup, 3600_000) rồi scale lên 3 instance, và cleanup chạy 3 lần.

Bốn cách chạy việc định kỳ:

Cách	Ưu	Nhược
`setInterval` trong app	Đơn giản nhất	Chạy trên mọi instance; mất khi restart; trôi thời gian
BullMQ repeatable job	Chỉ một instance chạy (Redis điều phối); có retry, có lịch sử	Phụ thuộc Redis; cần hiểu `jobId` deterministic
Cron hệ thống / K8s CronJob	Tách hẳn khỏi app; đúng một lần	Cần hạ tầng riêng; khó quan sát trong app
Managed scheduler (EventBridge, Cloud Scheduler)	Không vận hành gì	Khoá vào nhà cung cấp; gọi qua HTTP nên phải bảo vệ endpoint

Khuyến nghị cho DA3: BullMQ repeatable job — cùng hạ tầng, cùng retry, cùng observability với job thường.

await queue.upsertJobScheduler(
  'nightly-cleanup',                       // key ổn định → không sinh lịch trùng
  { pattern: '0 3 * * *', tz: 'Asia/Ho_Chi_Minh' },
  { name: 'cleanup', data: {} },
)

Pitfall #6 — lịch trùng lặp. Với API repeatable cũ, mỗi lần app khởi động lại mà repeatJobKey khác đi là bạn có thêm một lịch nữa. Sau 10 lần deploy, job chạy 10 lần mỗi đêm. Luôn dùng key ổn định (upsertJobScheduler), và liệt kê lịch hiện có lúc boot để kiểm tra:

const schedulers = await queue.getJobSchedulers()
logger.info({ count: schedulers.length, keys: schedulers.map(s => s.key) }, 'cron schedulers')

Múi giờ. Luôn khai báo tz tường minh. Server chạy UTC, người dùng ở Asia/Ho_Chi_Minh; "báo cáo hàng ngày lúc 3 giờ sáng" là 3 giờ sáng của ai? Và cron 0 2 * * * ở múi giờ có DST có thể bị bỏ hoặc chạy hai lần vào ngày chuyển giờ. Chi tiết → GĐ13 mục về timezone.

8.1 Leader election — chạy đúng một lần trên N instance

Khi không dùng được BullMQ (ví dụ việc phải chạy trong process API), cần khoá phân tán:

// Khoá Redis đơn giản — đủ dùng cho việc không quan trọng sống chết
async function withLock(key: string, ttlMs: number, fn: () => Promise<void>) {
  const token = crypto.randomUUID()
  const ok = await redis.set(key, token, 'PX', ttlMs, 'NX')   // NX = chỉ đặt nếu chưa có
  if (!ok) return                                             // instance khác đang giữ
  try {
    await fn()
  } finally {
    // Chỉ xoá nếu token vẫn là của mình — tránh xoá nhầm khoá của instance khác
    await redis.eval(
      `if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`,
      1, key, token,
    )
  }
}
await withLock('cron:cleanup', 5 * 60_000, cleanup)

Cảnh báo trung thực về khoá Redis. Cơ chế trên không an toàn tuyệt đối: nếu tiến trình giữ khoá bị treo (GC pause, network partition) quá TTL, instance khác lấy được khoá trong khi tiến trình đầu vẫn tưởng mình đang giữ → hai bên cùng chạy. Đây là vấn đề nổi tiếng (tranh luận Redlock giữa Martin Kleppmann và Salvatore Sanfilippo, 2016) và không có cách sửa hoàn toàn ở tầng khoá.

Cách xử lý đúng: đừng dựa vào khoá để đảm bảo tính đúng đắn. Dùng khoá để giảm việc trùng (tối ưu hoá), và dựa vào idempotency để đảm bảo đúng đắn. Nếu thực sự cần đúng-một-lần cứng, dùng hệ thống có consensus thật: etcd, ZooKeeper, hoặc lease của Kubernetes (→ GĐ16 mục về consensus).

Cách đơn giản nhất và thường là đúng nhất: dùng khoá ở tầng DB, đúng ngữ nghĩa:

sql

-- Advisory lock của Postgres: tự nhả khi session kết thúc, kể cả khi process chết
SELECT pg_try_advisory_lock(hashtext('cron:cleanup'));

8.2 Missed run và catch-up

Server sập từ 2:00 đến 4:00, cron 3:00 không chạy. Chuyện gì xảy ra khi server lên?

Ba chính sách, chọn có chủ đích:

Bỏ qua — hợp với việc idempotent chạy thường xuyên (dọn cache mỗi giờ). Mặc định của BullMQ.
Chạy bù một lần — hợp với việc tổng hợp (báo cáo ngày).
Chạy bù tất cả — hầu như luôn sai; sập 2 ngày là 48 lần chạy dồn, gây quá tải ngay lúc vừa hồi phục.

Thiết kế tốt hơn cron thuần: job theo khoảng dữ liệu, không theo thời điểm. Thay vì "chạy lúc 3h sáng để xử lý hôm qua", hãy lưu con trỏ:

// Job biết mình đã xử lý tới đâu → tự bù, chạy lại vô hại, không phụ thuộc lịch
const { lastProcessedAt } = await db.jobCursor.findUniqueOrThrow({ where: { name: 'daily-rollup' } })
const until = startOfDay(new Date())
for (const day of eachDayBetween(lastProcessedAt, until)) {
  await rollupDay(day)                                        // idempotent
  await db.jobCursor.update({ where: { name: 'daily-rollup' }, data: { lastProcessedAt: day } })
}

Thiết kế này tự động xử lý missed run, chạy lại được bao nhiêu lần cũng được, và test được mà không cần giả lập thời gian.

9. Fairness: không để một tenant chiếm cả queue

Vấn đề "noisy neighbor" phiên bản queue. Tenant A upload 50.000 file. Queue có 50.000 job của A. Tenant B upload 1 file — nằm ở vị trí 50.001 và chờ 4 tiếng.

Ba cách xử lý, từ đơn giản tới phức tạp:

Queue riêng theo mức ưu tiên: emails:critical (OTP, reset password) và emails:bulk (newsletter). Worker riêng cho mỗi queue. Đơn giản, hiệu quả, làm trước tiên.
priority của BullMQ: số nhỏ chạy trước. Nhưng cẩn thận — priority queue có thể gây starvation: job priority thấp không bao giờ tới lượt nếu job priority cao liên tục đổ vào.
Rate limit theo nhóm — đúng nhất cho multi-tenant:

const worker = new Worker('emails', handler, { connection, concurrency: 20 })

// Mỗi tenant tối đa 10 job / phút, phần dư tự động đẩy sang sau
await queue.add('send', payload, { group: { id: `tenant:${tenantId}` } })
// (BullMQ Pro có group rate limit; bản OSS: tự chia queue theo tenant hoặc
//  dùng token bucket trong Redis kiểm tra ở đầu handler rồi re-enqueue có delay)

Cách thủ công dùng được với BullMQ OSS:

async function handler(job: Job) {
  const allowed = await tokenBucket.take(`tenant:${job.data.tenantId}`, 10, 60_000)
  if (!allowed) {
    await queue.add(job.name, job.data, { delay: 10_000 })   // trả lại queue, thử sau
    return
  }
  await doWork(job.data)
}

10. Test job và worker

Nguyên tắc: tách handler khỏi hạ tầng queue. Handler là một hàm thuần nhận payload — test nó trực tiếp, không cần Redis.

// handler là hàm độc lập, không biết BullMQ tồn tại
export async function handleSendEmail(payload: SendEmailPayload, deps: Deps) { ... }

// worker.ts chỉ nối dây
new Worker('emails', (job) => handleSendEmail(job.data, deps), { connection })

// test — không Redis, không worker
it('không gửi lại email đã gửi', async () => {
  await handleSendEmail(payload, deps)
  await handleSendEmail(payload, deps)          // chạy lần hai
  expect(deps.mailer.send).toHaveBeenCalledTimes(1)   // idempotent
})

Test tích hợp queue (Redis thật qua Testcontainers → GĐ12): kiểm tra job được đẩy vào queue với payload đúng, kiểm tra retry, kiểm tra đường vào DLQ.

Không dùng sleep cố định để chờ job xong. Dùng event của BullMQ hoặc polling có timeout:

await new Promise<void>((resolve, reject) => {
  const t = setTimeout(() => reject(new Error('job timeout')), 5_000)
  new QueueEvents('emails', { connection }).on('completed', () => { clearTimeout(t); resolve() })
})

Test cron bằng cách gọi thẳng hàm, không chờ lịch. Với logic phụ thuộc thời gian, tiêm clock thay vì gọi new Date() trực tiếp — cùng nguyên tắc như test timezone ở GĐ13.

11. Bài tập — nâng cấp DA3

Yêu cầu.

Tách worker thành process riêng, Dockerfile chung nhưng command khác; docker-compose chạy api + worker + redis + postgres.
Ba queue: critical (OTP, reset password), default (email giao dịch), bulk (newsletter, export) — worker và concurrency riêng cho mỗi queue.
Outbox cho mọi sự kiện quan trọng; relay chạy trong worker process với SKIP LOCKED; cron dọn outbox cũ hơn 7 ngày.
Handler idempotent bằng bảng processed_job có UNIQUE; chứng minh bằng test gọi handler hai lần.
Retry có exponential backoff + jitter; phân biệt lỗi tạm thời và vĩnh viễn.
DLQ + log có cấu trúc + endpoint /admin/dlq (sau xác thực) để xem và chạy lại.
Graceful shutdown xử lý SIGTERM, chờ job đang chạy xong; ghi lại ba con số (job dài nhất / grace period / lockDuration) vào README.
Cron: dọn dẹp hàng đêm (BullMQ scheduler, tz tường minh) + một job rollup theo con trỏ như mục 8.2.
Metric: queue depth, oldest job age, failure rate, DLQ size — xuất ra /metrics hoặc log; đặt một cảnh báo thật.
Load test: đẩy 10.000 job, đo throughput và queue depth theo thời gian; giết worker giữa chừng và chứng minh không mất job, không xử lý trùng.

Mục 10 là sản phẩm giao quan trọng nhất. "Tôi đã giết worker giữa lúc chạy 10.000 job và chứng minh được không mất, không trùng" là một câu chuyện mạnh hơn mọi mô tả kiến trúc.

Done khi

[ ] Giải thích được 3 lợi ích của xử lý nền và cái giá phải trả
[ ] Phân biệt queue vs log (BullMQ vs Kafka); biết khi nào cần replay
[ ] Giải thích vì sao exactly-once không tồn tại và đạt hiệu quả đó bằng idempotency
[ ] Viết được handler idempotent bằng cả 3 cách ở mục 3
[ ] Giải thích outbox và vì sao queue.add() trong transaction sai theo cả hai chiều
[ ] Hiểu FOR UPDATE SKIP LOCKED giải quyết gì và vì sao nó cho phép nhiều relay
[ ] Payload nhỏ, có v; biết vì sao không nhét cả object và không để PII trong Redis
[ ] Retry có backoff + jitter; giải thích thundering herd
[ ] Phân biệt lỗi tạm thời / vĩnh viễn; không retry lỗi vĩnh viễn
[ ] Biết quan hệ job time < grace period < lockDuration và hậu quả khi sai
[ ] Worker chạy process riêng, xử lý SIGTERM, chờ job đang chạy xong
[ ] Tính được concurrency tối đa từ connection pool
[ ] Có DLQ; job không bao giờ biến mất im lặng; chạy lại là hành động có chủ đích
[ ] Theo dõi queue depth và oldest job age, không chỉ failure rate
[ ] Chạy cron đúng một lần trên N instance; biết giới hạn của khoá Redis và vì sao vẫn cần idempotency
[ ] Cron có tz tường minh; xử lý được missed run
[ ] Thiết kế được job theo con trỏ dữ liệu thay vì theo thời điểm
[ ] Chống được noisy neighbor bằng queue riêng hoặc rate limit theo tenant
[ ] Test handler không cần Redis; test tích hợp không dùng sleep cố định

Câu hỏi mở / chưa giải quyết

Khi nào chuyển từ BullMQ sang Kafka? Tín hiệu: cần replay lịch sử, cần nhiều consumer group độc lập trên cùng luồng sự kiện, hoặc throughput vượt khả năng của một Redis. Không phải "vì hệ thống lớn".
pg-boss có đủ thay BullMQ không? Với tải vừa và khi bạn đã có Postgres, có — và bạn được thêm lợi ích lớn là enqueue trong cùng transaction với dữ liệu, tức là không cần outbox. Đánh đổi: throughput thấp hơn Redis và tạo tải ghi lên DB chính.
Saga cho quy trình dài nhiều bước (đặt hàng → thanh toán → kho → giao hàng, có bồi hoàn khi hỏng giữa chừng) thuộc GĐ17, không thuộc giai đoạn này.

GĐ9 — Queues, Jobs, Workers & Cronjob: xử lý nền đúng cách ​

1. Vì sao cần xử lý nền ​

2. Chọn công nghệ hàng đợi ​

3. Đảm bảo giao nhận: at-most-once / at-least-once / exactly-once ​

4. Transactional Outbox — bài toán hai-lần-ghi ​

5. Thiết kế job: payload, retry, timeout ​

6. Dead Letter Queue và quan sát ​

7. Worker: vòng đời và graceful shutdown ​

8. Cronjob và scheduled job ​

8.1 Leader election — chạy đúng một lần trên N instance ​

8.2 Missed run và catch-up ​

9. Fairness: không để một tenant chiếm cả queue ​

10. Test job và worker ​

11. Bài tập — nâng cấp DA3 ​

Done khi ​

Câu hỏi mở / chưa giải quyết ​