PostgreSQL SKIP LOCKED thực chiến: Xây hàng đợi job nhiều worker

Nhiều worker cùng đọc một bảng job rất dễ lấy trùng công việc hoặc xếp hàng chờ cùng một row. PostgreSQL giải quyết bài toán claim song song bằng FOR UPDATE SKIP LOCKED: mỗi worker khóa một nhóm job sẵn sàng, bỏ qua job đã bị worker khác giữ, rồi cập nhật trạng thái và nhận dữ liệu bằng một câu lệnh nguyên tử.

Bài viết xây dựng một queue thực dụng với claim theo batch, retry có lịch, lease, heartbeat, run token, idempotency, index và metric. Theo tài liệu SELECT chính thức, SKIP LOCKED tạo view không nhất quán nên không phù hợp cho truy vấn thông thường, nhưng hữu ích để tránh tranh chấp giữa nhiều consumer của một bảng dạng queue. Phần khóa row và trả dữ liệu thay đổi dựa trên tài liệu về row-level lock và RETURNING.

1. Race condition của SELECT rồi UPDATE

SELECT id, payload
FROM jobs
WHERE status = 'pending'
ORDER BY id
LIMIT 1;

UPDATE jobs
SET status = 'running'
WHERE id = :id;

Nếu hai worker chạy SELECT trước khi bên nào UPDATE, cả hai có thể nhận cùng ID. Thêm FOR UPDATE trong transaction ngăn lấy trùng, nhưng worker sau sẽ chờ row đầu tiên dù phía sau còn hàng nghìn job rảnh. SKIP LOCKED thay hành vi chờ bằng bỏ qua row đang bị giữ để tìm job kế tiếp.

Đây là cơ chế phân phối công việc, không phải thứ tự FIFO tuyệt đối. Khi một row bị khóa, worker khác có quyền đi tiếp.

2. Schema queue có trạng thái và lease

CREATE TABLE jobs (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    queue_name text NOT NULL,
    payload jsonb NOT NULL,
    status text NOT NULL DEFAULT 'pending',
    priority smallint NOT NULL DEFAULT 0,
    available_at timestamptz NOT NULL DEFAULT now(),
    attempts integer NOT NULL DEFAULT 0,
    max_attempts integer NOT NULL DEFAULT 5,
    locked_at timestamptz,
    locked_by text,
    run_token uuid,
    last_error text,
    created_at timestamptz NOT NULL DEFAULT now(),
    updated_at timestamptz NOT NULL DEFAULT now(),

    CONSTRAINT jobs_status_valid
        CHECK (status IN ('pending', 'running', 'succeeded', 'failed')),
    CONSTRAINT jobs_attempts_valid
        CHECK (attempts >= 0 AND max_attempts > 0)
);

CREATE INDEX jobs_claim_idx
ON jobs (queue_name, priority DESC, available_at, id)
WHERE status = 'pending';

Không đưa now() vào predicate của partial index vì điều kiện index cần ổn định. Index thu hẹp theo status = 'pending'; query vẫn lọc available_at <= clock_timestamp() lúc chạy. Thứ tự có thêm id để ổn định khi priority và thời điểm bằng nhau.

3. Claim job bằng một câu lệnh nguyên tử

WITH picked AS (
    SELECT id
    FROM jobs
    WHERE queue_name = $1
      AND status = 'pending'
      AND available_at <= clock_timestamp()
      AND attempts < max_attempts
    ORDER BY priority DESC, available_at, id
    FOR UPDATE SKIP LOCKED
    LIMIT $2
)
UPDATE jobs AS j
SET status = 'running',
    attempts = j.attempts + 1,
    locked_at = clock_timestamp(),
    locked_by = $3,
    run_token = gen_random_uuid(),
    updated_at = clock_timestamp()
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.payload, j.attempts, j.run_token;

CTE chọn và khóa tối đa $2 row. UPDATE đổi quyền sở hữu ngay trong cùng statement; RETURNING đưa payload và token về worker mà không cần query thứ hai. gen_random_uuid() có trên các PostgreSQL hiện đại; nếu phiên bản hoặc policy khác, ứng dụng có thể truyền token ngẫu nhiên chất lượng cao.

Transaction chỉ nên bao quanh claim và commit ngay. Không giữ row lock trong suốt thời gian gọi API, gửi email hoặc xử lý video; transaction dài gây bloat, giữ connection và làm recovery khó hơn.

4. Vòng đời worker đúng cách

1. Mở transaction ngắn và chạy câu claim.
2. Commit; nếu không có row, chờ có jitter rồi poll lại.
3. Xử lý mỗi job ngoài transaction claim.
4. Định kỳ heartbeat nếu job chạy lâu.
5. Hoàn tất bằng update có điều kiện theo id + run_token.
6. Nếu thất bại, lên lịch retry hoặc chuyển trạng thái failed.

Batch lớn giảm số round trip nhưng tăng thời gian job đã bị claim mà chưa được xử lý. Bắt đầu với batch nhỏ, đo thời gian xử lý và điều chỉnh theo workload thay vì đặt một con số chung cho mọi queue.

5. Run token ngăn stale worker ghi kết quả

UPDATE jobs
SET status = 'succeeded',
    locked_at = NULL,
    locked_by = NULL,
    run_token = NULL,
    updated_at = clock_timestamp()
WHERE id = $1
  AND status = 'running'
  AND run_token = $2
RETURNING id;

Giả sử worker A bị treo, reaper đưa job về pending và worker B claim lại. A tỉnh dậy sau đó không được phép đánh dấu kết quả của B là succeeded. Token mới cho mỗi lần claim đóng vai trò fencing ở cấp bản ghi: update trả 0 row nghĩa là A đã mất lease và phải bỏ kết quả.

locked_by chỉ phục vụ quan sát; hostname hoặc process ID có thể tái sử dụng nên không đủ làm fencing token.

6. Heartbeat và reaper cho worker chết

-- Worker gia hạn lease
UPDATE jobs
SET locked_at = clock_timestamp(),
    updated_at = clock_timestamp()
WHERE id = $1
  AND status = 'running'
  AND run_token = $2;

-- Reaper trả job quá hạn về queue
UPDATE jobs
SET status = 'pending',
    available_at = clock_timestamp() + interval '5 seconds',
    locked_at = NULL,
    locked_by = NULL,
    run_token = NULL,
    updated_at = clock_timestamp()
WHERE status = 'running'
  AND locked_at < clock_timestamp() - interval '2 minutes'
  AND attempts < max_attempts;

Lease timeout phải lớn hơn heartbeat interval và chịu được pause thực tế của runtime, GC, network hoặc dependency. Reaper cũng cần chuyển job đã hết lần thử sang failed; nếu không, row có thể mắc kẹt ở running mãi mãi.

7. Retry có backoff và dead-letter

UPDATE jobs
SET status = CASE
        WHEN attempts >= max_attempts THEN 'failed'
        ELSE 'pending'
    END,
    available_at = CASE
        WHEN attempts >= max_attempts THEN available_at
        ELSE clock_timestamp() + ($3 * interval '1 second')
    END,
    last_error = left($4, 4000),
    locked_at = NULL,
    locked_by = NULL,
    run_token = NULL,
    updated_at = clock_timestamp()
WHERE id = $1
  AND status = 'running'
  AND run_token = $2
RETURNING status, attempts;

Ứng dụng tính delay exponential có jitter và giới hạn trần. Phân loại lỗi permanent như payload sai, địa chỉ không hợp lệ hoặc quyền bị thu hồi để fail ngay; chỉ retry lỗi có khả năng hồi phục. Bảng dead-letter riêng là lựa chọn tốt khi cần retention, quyền truy cập và workflow xử lý khác với queue đang chạy.

8. Queue này cung cấp at-least-once, không phải exactly-once

Worker có thể hoàn thành side effect bên ngoài rồi chết trước khi đánh dấu succeeded. Job sau đó được chạy lại. Không có transaction PostgreSQL nào tự động bao trùm email server, payment API hay object storage.

• Dùng idempotency key ổn định, thường là job ID hoặc business operation ID, khi gọi dependency hỗ trợ.
• Với thay đổi trong cùng database, ghi business state và trạng thái job trong một transaction.
• Với event sau commit, dùng transactional outbox và consumer deduplication.
• Thiết kế handler có thể chạy lại; không quảng cáo exactly-once khi chỉ có retry và row lock.

9. Ordering, priority và starvation

ORDER BY priority DESC, available_at, id biểu diễn ưu tiên, nhưng không hứa FIFO tuyệt đối vì row bị khóa sẽ bị bỏ qua. Luồng priority cao liên tục cũng có thể làm job thấp đói tài nguyên. Các biện pháp thực dụng gồm:

• Chia queue theo workload và cấp số worker tối thiểu cho mỗi queue.
• Age priority theo thời gian chờ hoặc dành một phần capacity cho priority thấp.
• Giới hạn số job từ cùng tenant trong một batch.
• Theo dõi tuổi của job pending lâu nhất, không chỉ queue depth.

Nếu nghiệp vụ yêu cầu thứ tự cứng theo account hoặc aggregate, serialize theo key đó hoặc dùng một sequencer chuyên biệt. SKIP LOCKED ưu tiên throughput hơn ordering tuyệt đối.

10. Isolation, lock và deadlock

READ COMMITTED thường là lựa chọn tự nhiên cho worker muốn thấy job vừa commit ở mỗi statement. Ở isolation cao hơn, ứng dụng phải sẵn sàng retry serialization failure theo contract của PostgreSQL. SKIP LOCKED chỉ bỏ qua row-level lock; statement vẫn lấy table-level lock thông thường và có thể chờ DDL xung đột.

Giữ thứ tự claim thống nhất và transaction ngắn để giảm deadlock. Không join tùy tiện nhiều bảng với locking clause nếu chưa xác định rõ bảng nào bị khóa; có thể dùng FOR UPDATE OF jobs để nói rõ mục tiêu.

11. Bloat, autovacuum và retention

Queue là bảng write-heavy: mỗi job đổi pending → running → succeeded/failed, cộng heartbeat và retry. Các update tạo row version chết; partial index cũng thay đổi khi status ra/vào pending. Production cần:

• Giữ payload vừa phải; dữ liệu lớn nên đặt ở object storage và queue chỉ lưu reference.
• Xóa hoặc archive job hoàn tất theo batch nhỏ và retention rõ ràng.
• Theo dõi dead tuples, autovacuum, table/index size, WAL và replica lag.
• Chỉnh autovacuum theo bảng queue nếu tốc độ churn cao.
• Dùng partition theo thời gian khi retention và volume thật sự biện minh độ phức tạp.

12. Đánh thức worker mà không biến NOTIFY thành queue

Producer có thể NOTIFY sau khi insert để worker đang ngủ poll sớm hơn. Nhưng notification chỉ là tín hiệu đánh thức, không phải nguồn dữ liệu bền vững: worker luôn phải query bảng và vẫn cần polling fallback. Payload, trạng thái, retry và quyền sở hữu nằm trong bảng job.

Khi insert job từ một business transaction, tạo job trong chính transaction đó để rollback cùng nghiệp vụ. Notification phát tại commit; worker không thấy công việc chưa commit.

13. Quan sát và kiểm thử concurrency

• Metric: pending/running/failed, claim rate, success rate, retry rate và lease expiry.
• Latency: queue wait, execution time và tuổi job pending lớn nhất theo queue/priority.
• Capacity: worker concurrency, database connections, rows claimed mỗi batch và query latency.
• Database: lock waits, deadlocks, dead tuples, WAL, index hit và slow plan.

Bài test quan trọng nhất mở nhiều connection đồng thời, claim hàng nghìn job và chứng minh mỗi lần claim có token riêng, không có hai active owner cho cùng row. Sau đó kill worker ở các điểm trước/sau side effect, chạy reaper, mô phỏng heartbeat trễ và xác minh stale completion bị từ chối.

Checklist production

• Claim là một statement nguyên tử có FOR UPDATE SKIP LOCKED và RETURNING.
• Transaction claim commit ngay; không giữ lock trong lúc xử lý job.
• Mỗi lần claim có run token; heartbeat và completion kiểm tra token.
• Lease, reaper, retry backoff, max attempts và failed/dead-letter được định nghĩa.
• Handler và side effect được thiết kế idempotent cho at-least-once delivery.
• Partial index khớp điều kiện pending và thứ tự claim.
• Đã có chính sách chống starvation, tenant fairness và batch size.
• Retention, autovacuum, bloat, WAL và replica lag được theo dõi.
• NOTIFY chỉ là wake-up hint; polling database vẫn là nguồn sự thật.
• Chaos test bao phủ crash, timeout, retry và stale worker.

Kết luận

FOR UPDATE SKIP LOCKED giúp PostgreSQL phân phối job hiệu quả giữa nhiều worker mà không tạo một hàng chờ khóa nối đuôi. Nhưng câu claim chỉ là phần đầu: một queue production còn cần transaction ngắn, run token, lease, retry, idempotency, fairness, retention và telemetry. Nếu chấp nhận semantics at-least-once và vận hành bảng write-heavy đúng cách, mô hình này là lựa chọn gọn, mạnh cho workload vừa và nhiều hệ thống muốn giữ business state cùng queue trong PostgreSQL.