Bài 3 — Thiết lập môi trường và bộ khung orchestrator

Bài này ta dựng bộ khung tối thiểu chạy được: một orchestrator đọc state từ DB, gọi các hệ thống qua adapter, và một queue để xử lý bất đồng bộ và retry. Ta dùng Python + PostgreSQL + Redis (hoặc bất kỳ queue nào) làm ví dụ, nhưng nguyên tắc áp dụng cho mọi stack.

Kiến trúc tổng quan

[API nhận đơn] --> ghi state PENDING --> đẩy job vào queue
                                              |
                     [Worker orchestrator] <--+
                        | đọc state
                        | gọi Adapter (WMS/Accounting/Payment)
                        | ghi state mới
                        | đẩy job kế tiếp

Tách rõ 3 lớp:

• Orchestrator core: biết state machine, quyết định bước kế.
• Adapters: mỗi hệ thống ngoài một adapter, che giấu chi tiết API.
• Queue/scheduler: đảm bảo job được chạy lại nếu worker chết.

Bước 1: Biến môi trường và kết nối

# .env
DATABASE_URL=postgres://user:pass@localhost:5432/orch
REDIS_URL=redis://localhost:6379/0
WMS_BASE_URL=https://wms.internal/api
PAYMENT_API_KEY=sk_test_xxx
MAX_ATTEMPTS=5

Đừng hardcode URL/khóa; orchestrator sẽ trỏ tới sandbox khi test, prod khi chạy thật.

Bước 2: Adapter chuẩn hóa

Mỗi adapter phơi ra một interface giống nhau: một hàm do() và một hàm undo() (bù trừ).

class PaymentAdapter:
    def do(self, ctx):
        # ctx chứa order_id, amount, idempotency_key
        resp = http.post(f"{BASE}/charge", json={
            "amount": ctx["amount"],
            "idempotency_key": ctx["idem_key"],  # bài 5 sẽ đào sâu
        }, headers={"Authorization": KEY})
        return {"payment_id": resp["id"]}

    def undo(self, ctx):
        http.post(f"{BASE}/refund", json={"payment_id": ctx["payment_id"]})

Interface đồng nhất cho phép orchestrator gọi bất kỳ bước nào mà không cần biết chi tiết.

Bước 3: Vòng lặp orchestrator

STEPS = [
    ("PENDING",  "RESERVED", wms_adapter),
    ("RESERVED", "INVOICED", acc_adapter),
    ("INVOICED", "PAID",     pay_adapter),
]

def advance(order_id):
    row = db.lock(order_id)
    for from_s, to_s, adapter in STEPS:
        if row.state == from_s:
            result = adapter.do(row.context())
            row.apply(to_s, result)
            db.save(row)
            queue.push(order_id)   # tiếp tục bước sau (bài 4 nói retry)
            return

Mỗi lần worker chỉ tiến một bước rồi lưu state — nhờ vậy nếu chết, lần sau đọc lên tiếp đúng chỗ.

Checklist thiết lập

• [ ] DB có bảng state với unique key theo order_id.
• [ ] Mỗi hệ thống ngoài có adapter với do()/undo().
• [ ] Có queue để re-dispatch job.
• [ ] Cấu hình qua biến môi trường, tách sandbox/prod.
• [ ] Có endpoint healthcheck cho worker.
• [ ] Log ghi kèm order_id để trace xuyên hệ thống.

Bước 4: Chạy thử end-to-end

• Trỏ mọi adapter về sandbox/mock.
• Tạo một đơn, quan sát state đi từ PENDING → PAID trong DB.
• Cố tình dừng worker giữa chừng, khởi động lại, xác nhận đơn chạy tiếp.
• Kiểm tra log có đủ order_id ở mọi dòng.

Vì sao tách API nhận đơn khỏi worker

Một lỗi kinh điển của người mới là gọi thẳng cả chuỗi WMS → Accounting → Payment ngay trong controller HTTP xử lý request tạo đơn. Vấn đề: request đó có thể mất 10–20 giây, dễ timeout ở tầng proxy/load balancer, và nếu client mất kết nối giữa chừng bạn không biết luồng đã chạy tới đâu. Thay vào đó, controller chỉ nên làm hai việc rất nhanh: (1) ghi state PENDING vào DB, (2) đẩy một job vào queue — rồi trả 202 Accepted kèm order_id cho client. Toàn bộ điều phối nặng chuyển sang worker chạy nền. Nhờ vậy request HTTP luôn nhanh và ổn định, còn quy trình dài được bảo vệ bởi state bền và queue.

# Controller nhận đơn — nhanh, không gọi hệ thống ngoài
def create_order(payload):
    order_id = new_order_id()
    db.insert("order_workflow", {
        "order_id": order_id,
        "state": "PENDING",
        "step_data": {"amount": payload["amount"]},
    })
    queue.push(order_id)          # worker sẽ tiếp quản
    return {"order_id": order_id, "status": "accepted"}, 202

Nguyên tắc: mỗi lần chạy tiến đúng một bước

Hãy để worker tiến một bước mỗi lần rồi lưu state và đẩy job kế tiếp, thay vì ôm cả chuỗi trong một transaction dài. Lý do: transaction dài giữ khóa lâu, khó retry riêng từng bước, và nếu chết ở bước cuối bạn phải chạy lại từ đầu. Với mô hình "một bước mỗi lần", điểm khôi phục (checkpoint) chính là các trạng thái trong DB — mịn và an toàn.

Sai lầm thường gặp

• Gọi thẳng API trong controller HTTP, không qua queue. Request timeout là mất luồng.
• Adapter không có undo(): tới bài compensation mới phát hiện không lùi được.
• Một worker tiến nhiều bước trong một transaction dài: giữ khóa lâu, khó retry từng phần.
• Dùng chung API key sandbox và prod: một hôm test lại charge tiền thật của khách.