Xây dựng hệ thống phân phối Webhook: Postback đáng tin cậy vào năm 2026

Cập nhật lần cuối vào ngày 24 tháng 2026 năm XNUMX bởi Caesar Fikson

Trả lời trực tiếp: Một dịch vụ gửi webhook hoặc postback liên kết đáng tin cậy cần có hàng đợi bền vững, số lần thử lại có giới hạn với độ trễ, cơ chế ngắt mạch cho từng điểm cuối, bảo vệ chống trùng lặp, xác minh chữ ký và dữ liệu đo lường cho thấy liệu quá trình gửi có bị chậm lại trước khi thất bại hay không.

Cách triển khai này cố tình giữ mô hình thực thi ở mức nhỏ: sử dụng các worker Python và PostgreSQL cho hàng đợi, lịch sử gửi và trạng thái hoạt động của điểm cuối. Đây là điểm khởi đầu thiết thực cho các doanh nghiệp SaaS B2B, hoạt động tiếp thị liên kết iGaming và bất kỳ sản phẩm nào cần gửi các sự kiện đi mà không coi mỗi lần hết thời gian chờ HTTP là một lần chuyển đổi bị mất.

Những điều mà hợp đồng giao hàng phải đảm bảo

Bảo mật và kiểm soát trùng lặp được ưu tiên hơn việc tinh chỉnh thử lại.

Hãy coi phần thân yêu cầu gửi đi là dữ liệu hoạt động nhạy cảm. Ký chính xác vào phần thân yêu cầu, bao gồm dấu thời gian gửi và ID sự kiện bất biến, xoay vòng các khóa bí mật ký và đảm bảo ứng dụng nhận có thể bỏ qua một cách an toàn việc phát lại cùng một sự kiện. Phản hồi HTTP thành công không phải là bằng chứng cho thấy một sự kiện nghiệp vụ đã được áp dụng chính xác một lần; phía nhận phải tự đưa ra quyết định đó.

Đối với quy trình postback của chương trình liên kết hoặc iGaming, hãy giữ ID nhấp chuột, ID chuyển đổi và các trường liên quan đến thanh toán ra khỏi nhật ký trừ khi các quy tắc kiểm soát truy cập và lưu giữ được nêu rõ. Tài liệu tham khảo Scaleo hiện có bên dưới chỉ liên quan như một ví dụ về độ trễ postback của chương trình liên kết; nó không thể thay thế cho việc lập tài liệu về hợp đồng phân phối giữa các dịch vụ của riêng bạn.

Các tài liệu tham khảo hữu ích về cách triển khai: Xác minh chữ ký webhook của Stripe, Lệnh SELECT và SKIP LOCKED trong PostgreSQLvà Hướng dẫn của AWS về cơ chế lùi lũy thừa và độ trễ.

Mô hình dữ liệu

Mọi thứ bắt đầu với hai bảng: một bảng dành cho hàng đợi giao hàng và một bảng dành cho các phép đo độ trễ.

sql

-- Pending and in-flight webhook deliveries
CREATE TABLE webhook_queue (
    id              SERIAL PRIMARY KEY,
    endpoint_id     INTEGER NOT NULL,
    endpoint_url    TEXT NOT NULL,
    payload         JSONB NOT NULL,
    
    -- Delivery state
    status          VARCHAR(20) NOT NULL DEFAULT 'pending',
        -- pending, in_flight, delivered, failed, dead_letter
    attempt_count   INTEGER NOT NULL DEFAULT 0,
    max_attempts    INTEGER NOT NULL DEFAULT 5,
    
    -- Timing
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    next_attempt_at TIMESTAMP NOT NULL DEFAULT NOW(),
    delivered_at    TIMESTAMP,
    last_error      TEXT,
    
    -- Response tracking
    last_status_code INTEGER,
    last_response_ms INTEGER,  -- response time in milliseconds
    
    INDEX idx_queue_next (status, next_attempt_at)
        WHERE status IN ('pending', 'failed')
);

-- Per-endpoint latency measurements (ring buffer)
CREATE TABLE endpoint_latency (
    id              SERIAL PRIMARY KEY,
    endpoint_id     INTEGER NOT NULL,
    response_ms     INTEGER NOT NULL,
    status_code     INTEGER,
    measured_at     TIMESTAMP NOT NULL DEFAULT NOW(),
    
    INDEX idx_latency_endpoint (endpoint_id, measured_at)
);

-- Endpoint health state (circuit breaker)
CREATE TABLE endpoint_health (
    endpoint_id         INTEGER PRIMARY KEY,
    endpoint_url        TEXT NOT NULL,
    state               VARCHAR(20) NOT NULL DEFAULT 'closed',
        -- closed (healthy), open (broken), half_open (testing)
    consecutive_failures INTEGER NOT NULL DEFAULT 0,
    failure_threshold   INTEGER NOT NULL DEFAULT 10,
    last_failure_at     TIMESTAMP,
    last_success_at     TIMESTAMP,
    opened_at           TIMESTAMP,  -- when circuit opened
    cooldown_seconds    INTEGER NOT NULL DEFAULT 300,  -- 5 min before half_open
    
    -- Latency stats (updated periodically)
    p50_ms              INTEGER,
    p95_ms              INTEGER,
    p99_ms              INTEGER,
    sample_count        INTEGER NOT NULL DEFAULT 0
);

Có ba điều cần lưu ý về lược đồ này.

Đầu tiên, webhook_queue bảng sử dụng một next_attempt_at cột thay vì một cơ chế lập lịch riêng biệt. Trình xử lý sẽ thăm dò các hàng trong đó status IN ('pending', 'failed') AND next_attempt_at <= NOW()Đây là một hệ thống xếp hàng chờ tạm thời đơn giản và hoạt động tốt với tốc độ khoảng 10,000 lượt gửi mỗi phút. Vượt quá tốc độ đó, hãy chuyển sang sử dụng một hệ thống môi giới tin nhắn chuyên nghiệp hơn.

Thứ hai, endpoint_latency Bảng này hoạt động như một bộ đệm vòng. Tôi định kỳ xóa các hàng cũ hơn 24 giờ. Tỷ lệ phần trăm độ trễ trong endpoint_health Các giá trị này được tính toán từ cửa sổ trượt này — chúng thể hiện hành vi gần đây, chứ không phải mức trung bình lịch sử.

Thứ ba, endpoint_health Bảng này thể hiện trạng thái máy của bộ ngắt mạch. Chi tiết hơn sẽ được trình bày bên dưới.

Người giao hàng

Vòng lặp xử lý cốt lõi được thiết kế đơn giản một cách có chủ ý. Sự phức tạp nằm ở logic thử lại và cơ chế ngắt mạch, chứ không phải ở chính đường dẫn phân phối.

mãng xà

import requests
import time
import psycopg2
from psycopg2.extras import RealDictCursor
from datetime import datetime, timedelta

DB_DSN = "postgresql://user:pass@localhost/webhooks"

def get_connection():
    return psycopg2.connect(DB_DSN)

def deliver_webhooks(batch_size=50):
    """
    Fetch pending webhooks and attempt delivery.
    Uses SELECT FOR UPDATE SKIP LOCKED for safe concurrent workers.
    """
    conn = get_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cur.execute("""
            SELECT id, endpoint_id, endpoint_url, payload, 
                   attempt_count, max_attempts
            FROM webhook_queue
            WHERE status IN ('pending', 'failed')
              AND next_attempt_at <= NOW()
            ORDER BY next_attempt_at ASC
            LIMIT %s
            FOR UPDATE SKIP LOCKED
        """, (batch_size,))
        
        rows = cur.fetchall()
        
        for row in rows:
            # Check circuit breaker before attempting
            if is_circuit_open(cur, row['endpoint_id']):
                # Don't attempt delivery — reschedule for after cooldown
                reschedule_for_cooldown(cur, row['id'], row['endpoint_id'])
                continue
            
            # Attempt delivery and measure latency
            result = attempt_delivery(
                row['endpoint_url'], 
                row['payload']
            )
            
            # Record latency measurement regardless of success/failure
            record_latency(
                cur, 
                row['endpoint_id'], 
                result['response_ms'], 
                result['status_code']
            )
            
            if result['success']:
                mark_delivered(cur, row['id'], result)
                record_success(cur, row['endpoint_id'])
            else:
                handle_failure(
                    cur, row['id'], row['endpoint_id'],
                    row['attempt_count'], row['max_attempts'],
                    result
                )
        
        conn.commit()
    
    except Exception as e:
        conn.rollback()
        raise
    finally:
        cur.close()
        conn.close()


def attempt_delivery(url, payload):
    """
    Fire the webhook and measure response time.
    Returns dict with success, status_code, response_ms, error.
    """
    start = time.monotonic()
    
    try:
        response = requests.post(
            url,
            json=payload,
            timeout=15,          # 15 second hard timeout
            headers={
                'Content-Type': 'application/json',
                'User-Agent': 'WebhookDelivery/1.0',
                'X-Delivery-Timestamp': str(int(time.time()))
            }
        )
        
        elapsed_ms = int((time.monotonic() - start) * 1000)
        
        return {
            'success': 200 <= response.status_code < 300,
            'status_code': response.status_code,
            'response_ms': elapsed_ms,
            'error': None if response.ok else f"HTTP {response.status_code}"
        }
    
    except requests.Timeout:
        elapsed_ms = int((time.monotonic() - start) * 1000)
        return {
            'success': False,
            'status_code': None,
            'response_ms': elapsed_ms,
            'error': 'timeout_15s'
        }
    
    except requests.ConnectionError as e:
        elapsed_ms = int((time.monotonic() - start) * 1000)
        return {
            'success': False,
            'status_code': None,
            'response_ms': elapsed_ms,
            'error': f'connection_error: {str(e)[:200]}'
        }

SELECT FOR UPDATE SKIP LOCKED Điều khoản này rất quan trọng để chạy nhiều phiên bản worker. Nếu không có nó, điều khoản này sẽ gây bất lợi. SKIP LOCKEDHai worker sẽ bị chặn trên cùng một hàng. Với cách này, mỗi worker sẽ lấy một nhóm webhook đang chờ xử lý khác nhau. Điều này cho phép bạn mở rộng quy mô theo chiều ngang bằng cách đơn giản là khởi động thêm các tiến trình worker.

time.monotonic() gọi thay vì time.time() Đó là hành động có chủ ý. time.time() có thể nhảy ngược lại trong quá trình điều chỉnh NTP. time.monotonic() Nó không bao giờ quay ngược lại, điều này rất quan trọng khi bạn đo độ trễ dưới một giây.

Logic thử lại với độ trễ lũy thừa và độ rung

Khi quá trình gửi dữ liệu thất bại, thời gian thử lại sẽ quyết định liệu hệ thống của bạn có phục hồi một cách trơn tru hay tạo ra một "đám đông" tấn công dồn dập vào điểm cuối đang gặp khó khăn.

mãng xà

import random

def calculate_next_attempt(attempt_count, base_delay=30, max_delay=3600):
    """
    Exponential backoff with full jitter.
    
    Attempt 1: 0-30s
    Attempt 2: 0-60s  
    Attempt 3: 0-120s
    Attempt 4: 0-240s
    Attempt 5: 0-480s (capped at max_delay)
    
    Full jitter prevents thundering herd when an endpoint
    recovers and hundreds of retries fire simultaneously.
    """
    exponential_delay = base_delay * (2 ** attempt_count)
    capped_delay = min(exponential_delay, max_delay)
    jittered_delay = random.uniform(0, capped_delay)
    
    return datetime.utcnow() + timedelta(seconds=jittered_delay)


def handle_failure(cur, webhook_id, endpoint_id, 
                   attempt_count, max_attempts, result):
    """
    Handle a failed delivery attempt.
    Either retry with backoff or move to dead letter queue.
    """
    new_attempt_count = attempt_count + 1
    
    if new_attempt_count >= max_attempts:
        # Exhausted retries — dead letter
        cur.execute("""
            UPDATE webhook_queue 
            SET status = 'dead_letter',
                attempt_count = %s,
                last_error = %s,
                last_status_code = %s,
                last_response_ms = %s
            WHERE id = %s
        """, (
            new_attempt_count, result['error'],
            result['status_code'], result['response_ms'],
            webhook_id
        ))
    else:
        # Schedule retry with backoff
        next_attempt = calculate_next_attempt(new_attempt_count)
        cur.execute("""
            UPDATE webhook_queue
            SET status = 'failed',
                attempt_count = %s,
                next_attempt_at = %s,
                last_error = %s,
                last_status_code = %s,
                last_response_ms = %s
            WHERE id = %s
        """, (
            new_attempt_count, next_attempt,
            result['error'], result['status_code'],
            result['response_ms'], webhook_id
        ))
    
    # Update circuit breaker
    record_failure(cur, endpoint_id)

Tại sao lại sử dụng jitter toàn phần thay vì jitter không tương quan hoặc jitter bằng nhau? AWS đã công bố phân tích chi tiết về vấn đề này. Jitter toàn phần (ngẫu nhiên hóa giữa 0 và giới hạn lũy thừa) tạo ra tổng thời gian hoàn thành thấp nhất trên tất cả các máy khách. Jitter bằng nhau (ngẫu nhiên hóa giữa một nửa giới hạn và giới hạn đầy đủ) thận trọng hơn nhưng chậm hơn trong việc xử lý hàng đợi thử lại. Đối với việc phân phối webhook, nơi bạn có nhiều điểm cuối độc lập, jitter toàn phần là lựa chọn đúng đắn vì mỗi lần thử lại của điểm cuối là độc lập — bạn không cần phối hợp giữa chúng.

Bộ ngắt mạch: Ngừng tấn công các điểm cuối bị hỏng

Mô hình ngắt mạch giúp hệ thống của bạn tránh lãng phí tài nguyên vào các điểm cuối liên tục gặp lỗi. Nếu không có nó, một điểm cuối bị lỗi sẽ tích lũy hàng trăm lần thử lại đang chờ xử lý, mỗi lần đều hết hạn sau 15 giây — làm tiêu hao năng lực xử lý của hệ thống vào những tác vụ sẽ không bao giờ thành công.

mãng xà

def is_circuit_open(cur, endpoint_id):
    """
    Check if the circuit breaker is open (endpoint is broken).
    If open and cooldown has passed, transition to half_open.
    """
    cur.execute("""
        SELECT state, opened_at, cooldown_seconds
        FROM endpoint_health
        WHERE endpoint_id = %s
    """, (endpoint_id,))
    
    row = cur.fetchone()
    if not row:
        return False  # no health record = assume healthy
    
    if row['state'] == 'closed':
        return False
    
    if row['state'] == 'open':
        # Check if cooldown period has elapsed
        if row['opened_at'] and row['cooldown_seconds']:
            elapsed = (datetime.utcnow() - row['opened_at']).total_seconds()
            if elapsed >= row['cooldown_seconds']:
                # Transition to half_open — allow one probe
                cur.execute("""
                    UPDATE endpoint_health
                    SET state = 'half_open'
                    WHERE endpoint_id = %s
                """, (endpoint_id,))
                return False  # allow the probe delivery
        return True  # still in cooldown
    
    if row['state'] == 'half_open':
        return False  # allow probe delivery
    
    return False


def record_failure(cur, endpoint_id):
    """
    Record a delivery failure. Open circuit if threshold reached.
    """
    cur.execute("""
        UPDATE endpoint_health
        SET consecutive_failures = consecutive_failures + 1,
            last_failure_at = NOW()
        WHERE endpoint_id = %s
        RETURNING consecutive_failures, failure_threshold, state
    """, (endpoint_id,))
    
    row = cur.fetchone()
    if not row:
        # Create health record on first failure
        cur.execute("""
            INSERT INTO endpoint_health (endpoint_id, endpoint_url, 
                consecutive_failures, last_failure_at)
            VALUES (%s, '', 1, NOW())
            ON CONFLICT (endpoint_id) DO UPDATE
            SET consecutive_failures = endpoint_health.consecutive_failures + 1,
                last_failure_at = NOW()
        """, (endpoint_id,))
        return
    
    if row['state'] == 'half_open':
        # Probe failed — re-open circuit with longer cooldown
        cur.execute("""
            UPDATE endpoint_health
            SET state = 'open',
                opened_at = NOW(),
                cooldown_seconds = LEAST(cooldown_seconds * 2, 3600)
            WHERE endpoint_id = %s
        """, (endpoint_id,))
    
    elif (row['state'] == 'closed' and 
          row['consecutive_failures'] >= row['failure_threshold']):
        # Threshold reached — open circuit
        cur.execute("""
            UPDATE endpoint_health
            SET state = 'open',
                opened_at = NOW(),
                cooldown_seconds = 300  -- reset to 5 minutes
            WHERE endpoint_id = %s
        """, (endpoint_id,))


def record_success(cur, endpoint_id):
    """
    Record a delivery success. Close circuit if half_open.
    """
    cur.execute("""
        UPDATE endpoint_health
        SET consecutive_failures = 0,
            last_success_at = NOW(),
            state = 'closed'
        WHERE endpoint_id = %s
    """, (endpoint_id,))

Chi tiết mà hầu hết các cách triển khai đều bỏ qua là cơ chế leo thang từ trạng thái half_open sang open với thời gian chờ gấp đôi. Nếu một endpoint bị lỗi trong quá trình thăm dò (trạng thái half_open), bạn không muốn thử lại sau 5 phút nữa. Endpoint đó vẫn bị lỗi. Hãy tăng gấp đôi thời gian chờ lên 10 phút, sau đó lên 20 phút, và tối đa là 1 giờ. Điều này ngăn chặn cơ chế ngắt mạch trở thành một cơ chế tấn công định kỳ.

Theo dõi phần trăm độ trễ

Số liệu trung bình không phản ánh chính xác thực tế. Một thiết bị đầu cuối với thời gian phản hồi trung bình là 200ms có thể phản hồi trong 50ms ở 95% thời gian và 3,000ms ở 5% thời gian còn lại. Con số trung bình trông có vẻ ổn. Nhưng P95 đã phát hiện ra một vấn đề ảnh hưởng đến 1 trong 20 lần gửi yêu cầu.

mãng xà

def record_latency(cur, endpoint_id, response_ms, status_code):
    """
    Record a latency measurement and update percentile stats.
    """
    cur.execute("""
        INSERT INTO endpoint_latency 
            (endpoint_id, response_ms, status_code)
        VALUES (%s, %s, %s)
    """, (endpoint_id, response_ms, status_code))


def update_latency_percentiles(cur, endpoint_id, window_hours=24):
    """
    Calculate P50, P95, P99 from the rolling window.
    Uses PostgreSQL's percentile_cont for exact percentiles.
    """
    cur.execute("""
        SELECT 
            COUNT(*) as sample_count,
            percentile_cont(0.50) WITHIN GROUP 
                (ORDER BY response_ms) AS p50,
            percentile_cont(0.95) WITHIN GROUP 
                (ORDER BY response_ms) AS p95,
            percentile_cont(0.99) WITHIN GROUP 
                (ORDER BY response_ms) AS p99
        FROM endpoint_latency
        WHERE endpoint_id = %s
          AND measured_at >= NOW() - INTERVAL '%s hours'
    """, (endpoint_id, window_hours))
    
    row = cur.fetchone()
    
    if row and row['sample_count'] > 0:
        cur.execute("""
            UPDATE endpoint_health
            SET p50_ms = %s,
                p95_ms = %s,
                p99_ms = %s,
                sample_count = %s
            WHERE endpoint_id = %s
        """, (
            int(row['p50']), int(row['p95']), 
            int(row['p99']), row['sample_count'],
            endpoint_id
        ))
    
    return row


def get_slow_endpoints(cur, p95_threshold_ms=2000):
    """
    Find endpoints whose P95 latency exceeds the threshold.
    These are candidates for investigation or circuit opening.
    """
    cur.execute("""
        SELECT endpoint_id, endpoint_url, 
               p50_ms, p95_ms, p99_ms, sample_count,
               state, consecutive_failures
        FROM endpoint_health
        WHERE p95_ms > %s
          AND sample_count >= 20  -- need sufficient samples
        ORDER BY p95_ms DESC
    """, (p95_threshold_ms,))
    
    return cur.fetchall()

PostgreSQL percentile_cont là một hàm tổng hợp tập hợp có thứ tự, tính toán các phân vị chính xác. Đối với các tập dữ liệu lớn, bạn sẽ chuyển sang sử dụng percentile_disc (trả về giá trị quan sát thực tế thay vì nội suy) hoặc sử dụng phép xấp xỉ t-digest. Đối với việc gửi webhook trong khoảng thời gian 24 giờ, các phân vị chính xác trên dữ liệu thô đủ nhanh cho đến khoảng 100,000 phép đo trên mỗi điểm cuối.

get_slow_endpoints Hàm này là hàm tôi chạy định kỳ để kiểm tra mỗi 15 phút. Các điểm cuối có P95 trên 2 giây sẽ được gắn cờ để điều tra. Các điểm cuối có P95 trên 5 giây sẽ bị giảm ngưỡng ngắt mạch — chúng được phép chịu ít lỗi liên tiếp hơn trước khi mạch mở, vì mỗi lần gửi không thành công sẽ chiếm dụng một luồng xử lý trong toàn bộ thời gian chờ.

Giám sát tình trạng hoạt động của Webhook

Đây là truy vấn giám sát mà tôi chạy cứ sau năm phút. Nó tạo ra một bản tóm tắt tình trạng hoạt động duy nhất của toàn bộ quy trình phân phối:

sql

SELECT
    -- Queue depth
    COUNT(*) FILTER (WHERE status = 'pending') AS pending,
    COUNT(*) FILTER (WHERE status = 'failed') AS awaiting_retry,
    COUNT(*) FILTER (WHERE status = 'in_flight') AS in_flight,
    COUNT(*) FILTER (WHERE status = 'dead_letter') AS dead_letter,
    
    -- Delivery rate (last hour)
    COUNT(*) FILTER (
        WHERE status = 'delivered' 
        AND delivered_at >= NOW() - INTERVAL '1 hour'
    ) AS delivered_last_hour,
    
    -- Failure rate (last hour)
    COUNT(*) FILTER (
        WHERE status IN ('failed', 'dead_letter')
        AND created_at >= NOW() - INTERVAL '1 hour'
    ) AS failed_last_hour,
    
    -- Oldest undelivered
    MIN(created_at) FILTER (
        WHERE status IN ('pending', 'failed')
    ) AS oldest_pending,
    
    -- Average delivery latency (last hour, successful only)
    AVG(last_response_ms) FILTER (
        WHERE status = 'delivered'
        AND delivered_at >= NOW() - INTERVAL '1 hour'
    ) AS avg_delivery_ms_last_hour

FROM webhook_queue;

oldest_pending Giá trị là chỉ số quan trọng nhất trong truy vấn này. Nếu nó cũ hơn cửa sổ thử lại tối đa của bạn (tổng của tất cả các độ trễ backoff), thì có điều gì đó không ổn về mặt cấu trúc — hoặc trình xử lý bị kẹt, điểm cuối bị chặn, hoặc hàng đợi đang tăng nhanh hơn tốc độ bạn có thể xử lý.

Tôi đưa ra cảnh báo khi có ba điều kiện: số lượng thư chết tăng lên (các điểm cuối liên tục bị lỗi và không ai điều tra), thời gian chờ trong hàng đợi vượt quá 30 phút (việc gửi thư đang bị chậm trễ), và Độ trễ P95 trên mỗi điểm cuối vượt quá ngưỡng cho thấy độ tin cậy phân phối bị suy giảm.Thứ ba là tín hiệu cảnh báo sớm — độ trễ tăng lên trước khi xảy ra lỗi. Một thiết bị đầu cuối trước đây phản hồi trong 200ms và bắt đầu phản hồi trong 3 giây thì sắp sửa bị hết thời gian chờ.

Hàng đợi thư chết không chỉ đơn thuần là nơi lưu trữ.

Hầu hết các nhóm đều triển khai hàng đợi thư chết (dead letter queue) như một bảng nơi các webhook bị lỗi được chuyển đến và bị loại bỏ. Họ chỉ kiểm tra nó định kỳ trong quá trình xử lý sự cố. Điều này thật lãng phí.

Hàng đợi thư chết là tập dữ liệu gỡ lỗi quý giá nhất của bạn. Mỗi hàng đại diện cho một lần gửi thư mà hệ thống đã thử nhiều lần và bỏ cuộc. Mô hình của các thư chết cho bạn biết những điều mà các chỉ số thành công không bao giờ tiết lộ.

mãng xà

def analyze_dead_letters(cur, hours=24):
    """
    Analyze recent dead letter entries for patterns.
    Returns per-endpoint failure analysis.
    """
    cur.execute("""
        SELECT 
            endpoint_id,
            endpoint_url,
            COUNT(*) AS dead_count,
            
            -- Most common error
            MODE() WITHIN GROUP (ORDER BY last_error) AS primary_error,
            
            -- Most common status code
            MODE() WITHIN GROUP (ORDER BY last_status_code) 
                AS primary_status_code,
            
            -- Timing
            MIN(created_at) AS first_dead,
            MAX(created_at) AS last_dead,
            
            -- Average attempts before giving up
            AVG(attempt_count)::INTEGER AS avg_attempts
            
        FROM webhook_queue
        WHERE status = 'dead_letter'
          AND created_at >= NOW() - INTERVAL '%s hours'
        GROUP BY endpoint_id, endpoint_url
        ORDER BY dead_count DESC
        LIMIT 20
    """, (hours,))
    
    return cur.fetchall()

Khi xem xét các thư không được sử dụng, tôi tìm kiếm ba mẫu sau.

Lỗi cụm: 50 thư chết gửi đến cùng một điểm cuối trong cùng một giờ có nghĩa là điểm cuối đó đã gặp sự cố và không khôi phục được trong khoảng thời gian cho phép thử lại. Hành động: kéo dài khoảng thời gian thử lại hoặc thực hiện xếp hàng lại thủ công.

Các mẫu mã trạng thái: Sự gia tăng đột biến các thư không hoạt động (401/403) có nghĩa là điểm cuối đã thay đổi thông tin xác thực và không ai cập nhật cấu hình webhook. Sự gia tăng đột biến lỗi 429 (Quá nhiều yêu cầu) có nghĩa là bạn đang vượt quá giới hạn tỷ lệ yêu cầu và cần phải điều chỉnh tốc độ.

Tích lũy dần dần: 2-3 thư chết mỗi ngày cho một điểm cuối duy nhất, phân bố đều. Đây là kiểu lỗi khó phát hiện nhất — điểm cuối hoạt động hầu hết thời gian nhưng gặp lỗi gián đoạn khiến số lần thử lại cạn kiệt theo thời gian. Cách khắc phục thường là tăng số lần thử lại. max_attempts cho điểm cuối cụ thể đó hoặc giảm thời gian chờ.

Vận hành công nhân

Vòng lặp chính liên kết mọi thứ lại với nhau:

mãng xà

import signal
import sys

running = True

def shutdown_handler(signum, frame):
    global running
    running = False
    print(f"Received signal {signum}, shutting down gracefully...")

signal.signal(signal.SIGTERM, shutdown_handler)
signal.signal(signal.SIGINT, shutdown_handler)

def main():
    print("Webhook delivery worker starting...")
    
    while running:
        try:
            deliver_webhooks(batch_size=50)
        except Exception as e:
            print(f"Worker error: {e}")
            time.sleep(5)  # back off on errors
            continue
        
        # Update latency stats every 100 iterations
        # (cheap operation, doesn't need to run every loop)
        if int(time.time()) % 100 == 0:
            conn = get_connection()
            cur = conn.cursor(cursor_factory=RealDictCursor)
            try:
                cur.execute(
                    "SELECT DISTINCT endpoint_id FROM endpoint_health"
                )
                for row in cur.fetchall():
                    update_latency_percentiles(cur, row['endpoint_id'])
                conn.commit()
            finally:
                cur.close()
                conn.close()
        
        # Poll interval — 500ms keeps latency low without
        # hammering the database
        time.sleep(0.5)

    print("Worker shut down cleanly.")

if __name__ == '__main__':
    main()

SIGTERM Trình xử lý này rất cần thiết cho việc tắt máy sạch sẽ trong môi trường container. Khi Kubernetes gửi tín hiệu SIGTERM, worker sẽ hoàn thành lô xử lý hiện tại, cam kết giao dịch và thoát. Nếu không có điều này, bạn sẽ gặp tình trạng các hàng bị kẹt trong quá trình xử lý. in_flight Trạng thái không có nhân viên nào xử lý chúng.

Những điều hệ thống này không làm được (và khi nào bạn cần thêm chức năng này)

Cách triển khai này xử lý tối đa khoảng 10,000 đơn hàng mỗi phút trên một phiên bản PostgreSQL duy nhất với 2-3 tiến trình xử lý. Vượt quá con số đó, cần thực hiện ba thay đổi.

Đầu tiên, hãy thay thế hàng đợi PostgreSQL bằng Redis Streams hoặc RabbitMQ. SELECT FOR UPDATE SKIP LOCKED Mô hình này tạo ra sự tranh chấp ghi trên bảng hàng đợi ở thông lượng cao. Một bộ điều phối tin nhắn chuyên dụng sẽ loại bỏ điều này.

Thứ hai, hãy thêm giới hạn tốc độ cho từng điểm cuối. Một số điểm cuối nhận yêu cầu có giới hạn tốc độ (100 yêu cầu mỗi phút, 1,000 yêu cầu mỗi giờ). Nếu không có giới hạn tốc độ phía máy khách, bạn sẽ vượt quá hạn mức của họ và nhận được lỗi 429. Hãy triển khai một token bucket cho mỗi điểm cuối.

Thứ ba, thêm chữ ký yêu cầu. Chữ ký HMAC-SHA256 trên dữ liệu gửi đi cho phép điểm cuối nhận xác minh rằng webhook đến từ hệ thống của bạn và không bị giả mạo trong quá trình truyền tải. Đây là điều kiện tiên quyết đối với bất kỳ hệ thống webhook nào gửi dữ liệu tài chính.

Hệ thống được đề cập trong bài viết này là nền tảng. Nó xử lý các vấn đề khó khăn — logic thử lại, ngắt mạch, đo độ trễ, phân tích thư chết — mà mọi hệ thống phân phối webhook cần bất kể quy mô. Các thành phần cụ thể mà bạn thêm vào (bộ điều phối tin nhắn, bộ giới hạn tốc độ, ký yêu cầu) phụ thuộc vào yêu cầu về thông lượng và bảo mật của bạn.

Phần quan trọng nhất lại là phần mà hầu hết các nhóm thường bỏ qua: đo lường chính hệ thống phân phối. Nếu bạn không thể trả lời câu hỏi “độ trễ phân phối P95 đến Điểm cuối X trong 24 giờ qua là bao nhiêu?”, thì bạn đang hoạt động một cách mù quáng. Hãy xây dựng hệ thống đo lường trước tiên. Mọi thứ khác sẽ theo sau.

Câu hỏi thường gặp về phân phối Webhook

Liệu trình gửi webhook có nên cam kết chỉ gửi một lần duy nhất?

Thông thường là không. Bên gửi nên hiển thị các lần thử lại và cung cấp ID sự kiện ổn định; bên nhận nên đảm bảo quá trình xử lý có tính chất bất biến để sự kiện có thể được gửi nhiều lần mà không làm trùng lặp kết quả nghiệp vụ.

Những lỗi nào cần được thử lại?

Chỉ thử lại các lỗi mà hợp đồng của bạn phân loại là tạm thời, chẳng hạn như lỗi mạng, hết thời gian chờ và phản hồi máy chủ được chọn. Không nên thử lại nhiều lần các yêu cầu không đúng định dạng, lỗi xác thực hoặc các lỗi vĩnh viễn khác mà không có lộ trình khắc phục rõ ràng.

Khi nào thì một nhóm nên chuyển sang sử dụng hệ thống xếp hàng thay vì chỉ dựa vào cơ sở dữ liệu?

Di chuyển khi các chỉ số đo lường như mức độ tranh chấp, thời gian tồn đọng, thông lượng hoặc nhu cầu phục hồi hoạt động cho thấy hàng đợi cơ sở dữ liệu không còn đáp ứng được hợp đồng cung cấp dịch vụ. Quyết định về dung lượng nên dựa trên khối lượng công việc thực tế, chứ không phải dựa trên tuyên bố chung chung về tốc độ yêu cầu.