การสร้างระบบส่งข้อมูลผ่าน Webhook: การส่งข้อมูลกลับที่เชื่อถือได้ในปี 2026

แก้ไขล่าสุดเมื่อ 24 มิถุนายน 2026 โดย ซีซาร์ ฟิกสัน

คำตอบตรงๆ : บริการส่ง webhook หรือ affiliate-postback ที่เชื่อถือได้ จำเป็นต้องมีคิวที่ทนทาน การลองส่งซ้ำที่จำกัดพร้อมค่าความคลาดเคลื่อนเล็กน้อย การตัดวงจรต่อปลายทาง การป้องกันการส่งข้อมูลซ้ำ การตรวจสอบลายเซ็น และระบบวัดระยะทางที่แสดงให้เห็นว่าการส่งช้าลงหรือไม่ก่อนที่จะล้มเหลว

การออกแบบนี้ตั้งใจให้โมเดลการทำงานมีขนาดเล็ก: ใช้ Python เป็นตัวประมวลผล และ PostgreSQL สำหรับคิว ประวัติการส่ง และสถานะของปลายทาง นี่คือจุดเริ่มต้นที่ใช้งานได้จริงสำหรับ B2B SaaS, การดำเนินงานพันธมิตร iGaming และผลิตภัณฑ์ใดๆ ที่ต้องส่งอีเวนต์ขาออกโดยไม่ทำให้การหมดเวลาของ HTTP เพียงครั้งเดียวกลายเป็นการสูญเสียการแปลง

สัญญาการส่งมอบต้องรับประกันอะไรบ้าง

การรักษาความปลอดภัยและการควบคุมการทำซ้ำมีความสำคัญมากกว่าการปรับแต่งการลองใหม่

ถือว่าส่วนเนื้อหาขาออก (outbound body) เป็นข้อมูลการดำเนินงานที่ละเอียดอ่อน ลงนามในส่วนเนื้อหาคำขออย่างถูกต้อง ระบุเวลาส่งมอบและรหัสเหตุการณ์ที่ไม่สามารถเปลี่ยนแปลงได้ หมุนเวียนรหัสลับการลงนาม และตรวจสอบให้แน่ใจว่าแอปพลิเคชันผู้รับสามารถเพิกเฉยต่อการเล่นซ้ำของเหตุการณ์เดียวกันได้อย่างปลอดภัย การตอบสนอง HTTP ที่สำเร็จไม่ได้เป็นหลักฐานว่าเหตุการณ์ทางธุรกิจถูกนำไปใช้เพียงครั้งเดียวเท่านั้น ฝั่งผู้รับต้องเป็นผู้ตัดสินใจในเรื่องนั้น

สำหรับเวิร์กโฟลว์การส่งข้อมูลกลับไปยังพันธมิตรหรือเกมออนไลน์ ควรหลีกเลี่ยงการบันทึกรหัสคลิก รหัสการแปลง และฟิลด์ที่เกี่ยวข้องกับการจ่ายเงิน เว้นแต่จะมีการกำหนดการควบคุมการเข้าถึงและกฎการเก็บรักษาไว้อย่างชัดเจน ตัวอย่างอ้างอิง Scaleo ด้านล่างนี้ใช้ได้เฉพาะเป็นตัวอย่างความหน่วงในการส่งข้อมูลกลับไปยังพันธมิตรเท่านั้น ไม่สามารถใช้แทนการจัดทำเอกสารสัญญาการส่งมอบระหว่างบริการของคุณได้

เอกสารอ้างอิงการนำไปใช้งานที่เป็นประโยชน์: การตรวจสอบลายเซ็น Stripe webhook, SELECT และ SKIP LOCKED ใน PostgreSQLและ คำแนะนำของ AWS เกี่ยวกับการหน่วงเวลาแบบทวีคูณและความผันผวนของเวลา.

แบบจำลองข้อมูล

ทุกอย่างเริ่มต้นด้วยตารางสองตาราง: ตารางหนึ่งสำหรับคิวการจัดส่ง และอีกตารางหนึ่งสำหรับการวัดค่าความหน่วง

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
);

มีสามสิ่งที่ควรสังเกตเกี่ยวกับโครงสร้างข้อมูลนี้

ครั้งแรกที่ webhook_queue ตารางใช้ next_attempt_at ใช้คอลัมน์แทนกลไกการจัดตารางเวลาแยกต่างหาก ตัวประมวลผลจะตรวจสอบแถวที่ตรงตามเงื่อนไข status IN ('pending', 'failed') AND next_attempt_at <= NOW()นี่คือระบบคิวหน่วงเวลาแบบประหยัด และใช้งานได้ดีจนถึงประมาณ 10,000 ข้อความต่อนาที หากเกินกว่านั้น ควรเปลี่ยนไปใช้ระบบตัวกลางส่งข้อความที่เหมาะสมกว่า

ประการที่สอง endpoint_latency ตารางนี้ทำหน้าที่เป็นบัฟเฟอร์แบบวงแหวน ฉันจะลบแถวที่เก่ากว่า 24 ชั่วโมงออกเป็นระยะๆ เปอร์เซ็นไทล์ของความหน่วงใน endpoint_health ตัวเลขเหล่านี้คำนวณจากช่วงเวลาหมุนเวียน — ซึ่งแสดงถึงพฤติกรรมล่าสุด ไม่ใช่ค่าเฉลี่ยในอดีต

ประการที่สาม endpoint_health ตารางนี้ใช้ในการสร้างเครื่องสถานะเบรกเกอร์วงจร รายละเอียดเพิ่มเติมอยู่ด้านล่าง

พนักงานส่งของ

วงจรการทำงานหลักถูกออกแบบมาให้เรียบง่ายโดยเจตนา ความซับซ้อนควรอยู่ที่ตรรกะการลองใหม่และกลไกตัดวงจร ไม่ใช่ที่เส้นทางการส่งมอบข้อมูลเอง

หลาม

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 เงื่อนไขนี้มีความสำคัญอย่างยิ่งสำหรับการเรียกใช้งานอินสแตนซ์ของเวิร์กเกอร์หลายตัว หากไม่มีเงื่อนไขนี้ SKIP LOCKEDโดยที่พนักงานสองคนจะบล็อกการทำงานในแถวเดียวกัน ด้วยวิธีนี้ พนักงานแต่ละคนจะรับชุดเว็บฮุคที่รอดำเนินการที่แตกต่างกัน ซึ่งจะช่วยให้คุณสามารถขยายขนาดในแนวนอนได้โดยการเริ่มต้นกระบวนการทำงานของพนักงานเพิ่มขึ้น

การขอ time.monotonic() โทรแทน time.time() เป็นการกระทำโดยเจตนา time.time() อาจกระโดดถอยหลังระหว่างการปรับ NTP time.monotonic() จะไม่ย้อนกลับ ซึ่งเป็นสิ่งสำคัญเมื่อคุณวัดความหน่วงเวลาในระดับต่ำกว่าหนึ่งวินาที

ตรรกะการลองใหม่ด้วยการหน่วงเวลาแบบเลขชี้กำลังและการกระเพื่อม

เมื่อการส่งล้มเหลว เวลาในการลองใหม่จะเป็นตัวกำหนดว่าระบบของคุณจะฟื้นตัวได้อย่างราบรื่นหรือจะสร้างปัญหาใหญ่ที่รุมเร้าปลายทางที่กำลังทำงานอยู่

หลาม

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)

ทำไมถึงเลือกใช้ค่า jitter เต็มรูปแบบ แทนที่จะใช้ค่า jitter ที่ไม่สัมพันธ์กัน หรือค่า jitter ที่เท่ากัน? AWS ได้เผยแพร่บทวิเคราะห์ที่ชัดเจนเกี่ยวกับเรื่องนี้แล้ว ค่า jitter เต็มรูปแบบ (การสุ่มค่าระหว่าง 0 ถึงค่าสูงสุดแบบเลขชี้กำลัง) จะทำให้เวลาการเสร็จสิ้นโดยรวมต่ำที่สุดสำหรับลูกค้าทุกราย ค่า jitter ที่เท่ากัน (การสุ่มค่าระหว่างครึ่งหนึ่งของค่าสูงสุดและค่าสูงสุด) นั้นระมัดระวังกว่า แต่จะใช้เวลานานขึ้นในการระบาย retry backlog สำหรับการส่ง webhook ที่มี endpoint อิสระจำนวนมาก ค่า jitter เต็มรูปแบบจึงเป็นตัวเลือกที่เหมาะสม เพราะการลองใหม่ของแต่ละ endpoint นั้นเป็นอิสระต่อกัน — คุณไม่ต้องประสานงานระหว่าง endpoint เหล่านั้น

เบรกเกอร์วงจร: หยุดการกระแทกจุดเชื่อมต่อที่ชำรุด

รูปแบบ Circuit Breaker ช่วยป้องกันไม่ให้ระบบของคุณสิ้นเปลืองทรัพยากรไปกับปลายทางที่ล้มเหลวอย่างต่อเนื่อง หากไม่มีรูปแบบนี้ ปลายทางที่ใช้งานไม่ได้จะสะสมการลองใหม่ที่ค้างอยู่หลายร้อยรายการ ซึ่งแต่ละรายการจะหมดเวลา 15 วินาที ทำให้สิ้นเปลืองกำลังการประมวลผลของ Worker ไปกับการส่งข้อมูลที่ไม่มีวันสำเร็จ

หลาม

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,))

การเพิ่มระดับสถานะจาก half_open เป็น open โดยเพิ่มระยะเวลา cooldown เป็นสองเท่า เป็นรายละเอียดที่การใช้งานส่วนใหญ่มองข้ามไป หากเอนด์พอยต์ล้มเหลวระหว่างการตรวจสอบ (สถานะ half_open) คุณไม่ต้องการลองใหม่ในอีก 5 นาที เพราะเอนด์พอยต์ยังคงเสียอยู่ ดังนั้นควรเพิ่มระยะเวลา cooldown เป็น 10 นาที จากนั้นเป็น 20 นาที และสูงสุดที่ 1 ชั่วโมง วิธีนี้จะช่วยป้องกันไม่ให้ circuit breaker กลายเป็นกลไกการทำงานที่ซ้ำซากและไม่หยุดหย่อน

การติดตามเปอร์เซ็นไทล์ความหน่วง

ค่าเฉลี่ยอาจไม่ตรงกับความเป็นจริง อุปกรณ์ปลายทางที่มีเวลาตอบสนองเฉลี่ย 200 มิลลิวินาที อาจตอบสนองภายใน 50 มิลลิวินาที 95% ของเวลา และตอบสนองภายใน 3,000 มิลลิวินาทีในอีก 5% ที่เหลือ ค่าเฉลี่ยดูเหมือนจะปกติ แต่ P95 เผยให้เห็นปัญหาที่ส่งผลกระทบต่อการส่งมอบ 1 ใน 20 ครั้ง

หลาม

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 เป็นฟังก์ชันการรวมชุดเรียงลำดับที่คำนวณเปอร์เซ็นไทล์ที่แม่นยำ สำหรับชุดข้อมูลขนาดใหญ่ คุณควรเปลี่ยนไปใช้ percentile_disc (ซึ่งจะส่งคืนค่าที่สังเกตได้จริง แทนที่จะเป็นการประมาณค่าแบบสอดแทรก) หรือใช้การประมาณค่าแบบ t-digest สำหรับการส่งข้อมูลผ่าน webhook ที่มีช่วงเวลา 24 ชั่วโมง การคำนวณเปอร์เซ็นไทล์ที่แม่นยำจากข้อมูลดิบนั้นเร็วพอสำหรับการวัดประมาณ 100,000 ครั้งต่อเอนด์พอยต์

การขอ get_slow_endpoints ฟังก์ชันนี้คือฟังก์ชันที่ผมใช้ตรวจสอบตามกำหนดเวลาทุกๆ 15 นาที เอนด์พอยต์ที่มีค่า P95 สูงกว่า 2 วินาที จะถูกตั้งค่าสถานะเพื่อตรวจสอบ ส่วนเอนด์พอยต์ที่มีค่า P95 สูงกว่า 5 วินาที จะถูกลดค่าเกณฑ์การตัดวงจรลง กล่าวคือ จะอนุญาตให้เกิดความล้มเหลวติดต่อกันได้น้อยลงก่อนที่วงจรจะเปิด เพราะการส่งข้อมูลที่ล้มเหลวแต่ละครั้งจะทำให้เธรดการทำงานถูกใช้งานจนครบระยะเวลาหมดเวลา

การตรวจสอบสถานะการทำงานของ Webhook

นี่คือคำสั่งตรวจสอบที่ผมรันทุกห้านาที คำสั่งนี้จะแสดงสรุปสถานะโดยรวมของกระบวนการส่งมอบทั้งหมดในแถวเดียว:

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 ค่าที่ได้คือตัวชี้วัดที่สำคัญที่สุดในการค้นหานี้ หากค่าที่ได้เก่ากว่าช่วงเวลาการลองใหม่สูงสุดของคุณ (ผลรวมของเวลาหน่วงทั้งหมด) แสดงว่ามีบางอย่างผิดปกติในโครงสร้าง — ไม่ว่าจะเป็นตัวประมวลผลค้าง ปลายทางถูกบล็อก หรือคิวเพิ่มขึ้นเร็วกว่าที่คุณจะระบายออกได้

ฉันจะแจ้งเตือนเมื่อเกิดเงื่อนไขสามประการ ได้แก่ จำนวนจดหมายที่ส่งไม่ถึงปลายทางเพิ่มขึ้น (ปลายทางล้มเหลวอย่างถาวรและไม่มีใครตรวจสอบ) อายุของคิวรอเกิน 30 นาที (การส่งล่าช้า) และ ค่าความหน่วง P95 ต่อจุดสิ้นสุดเกินเกณฑ์ที่บ่งชี้ถึงความน่าเชื่อถือในการส่งข้อมูลที่ลดลงประการที่สามคือสัญญาณเตือนล่วงหน้า — ความหน่วงจะเพิ่มขึ้นก่อนที่จะเกิดความล้มเหลว อุปกรณ์ปลายทางที่เคยตอบสนองภายใน 200 มิลลิวินาที และเริ่มตอบสนองใน 3 วินาที แสดงว่ากำลังจะเริ่มหมดเวลาแล้ว

คิวจดหมายที่ส่งไม่ถึงผู้รับไม่ใช่แค่ที่เก็บข้อมูล

ทีมส่วนใหญ่ใช้คิวข้อความที่ส่งไม่สำเร็จ (dead letter queue) เป็นตารางสำหรับเก็บ webhook ที่ล้มเหลว พวกเขาจะตรวจสอบคิวนี้เป็นครั้งคราวระหว่างการแก้ไขปัญหา ซึ่งเป็นการสิ้นเปลืองโดยเปล่าประโยชน์

คิวจดหมายที่ส่งไม่สำเร็จ (Dead Letter Queue) คือชุดข้อมูลการแก้ไขข้อผิดพลาดที่มีค่าที่สุดของคุณ แต่ละแถวแสดงถึงการส่งที่ระบบของคุณพยายามหลายครั้งแล้วแต่ก็ล้มเหลว รูปแบบของจดหมายที่ส่งไม่สำเร็จจะบอกสิ่งต่างๆ ที่ตัวชี้วัดความสำเร็จไม่สามารถบอกได้

หลาม

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()

เมื่อผมตรวจสอบจดหมายที่ส่งไม่ถึงผู้รับ ผมจะมองหารูปแบบสามอย่าง

ความล้มเหลวของคลัสเตอร์: การได้รับข้อความแจ้งเตือนการส่งไม่สำเร็จ (dead letters) จำนวน 50 ข้อความสำหรับปลายทางเดียวกันในชั่วโมงเดียวกัน หมายความว่าปลายทางนั้นล่มและไม่สามารถกู้คืนได้ภายในช่วงเวลาที่กำหนด การดำเนินการ: ขยายช่วงเวลาที่กำหนด หรือทำการจัดคิวใหม่ด้วยตนเอง

รูปแบบรหัสสถานะ: การพบข้อผิดพลาด 401/403 จำนวนมากผิดปกติ หมายความว่าปลายทางมีการเปลี่ยนข้อมูลรับรอง และไม่มีใครอัปเดตการตั้งค่าเว็บฮุค การพบข้อผิดพลาด 429 (Too Many Requests) จำนวนมากผิดปกติ หมายความว่าคุณกำลังใช้งานเกินขีดจำกัดอัตราการใช้งาน และจำเป็นต้องลดปริมาณการใช้งานลง

การสะสมทีละน้อย: มีการส่งจดหมายไม่สำเร็จ 2-3 ครั้งต่อวันสำหรับปลายทางเดียว โดยกระจายอย่างสม่ำเสมอ นี่คือรูปแบบที่ซ่อนเร้นที่สุด – ปลายทางส่วนใหญ่ทำงานได้ แต่มีข้อผิดพลาดเป็นระยะ ๆ ซึ่งทำให้จำนวนการลองใหม่หมดไปเมื่อเวลาผ่านไป การแก้ไขมักจะเป็นการเพิ่ม max_attempts สำหรับปลายทางเฉพาะนั้น หรือการลดระยะเวลาหมดเวลา

การใช้งานเครื่องทำงาน

วงจรหลักที่เชื่อมโยงทุกสิ่งเข้าด้วยกัน:

หลาม

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 ตัวจัดการสัญญาณ (handler) มีความสำคัญอย่างยิ่งสำหรับการปิดระบบอย่างราบรื่นในสภาพแวดล้อมคอนเทนเนอร์ เมื่อ Kubernetes ส่งสัญญาณ SIGTERM ตัวประมวลผลจะทำงานชุดปัจจุบันให้เสร็จสิ้น ยืนยันธุรกรรม และออกจากระบบ หากไม่มีตัวจัดการสัญญาณนี้ คุณจะพบแถวข้อมูลที่ค้างอยู่ in_flight สถานะที่ไม่มีพนักงานประมวลผลอยู่

ระบบนี้ทำอะไรไม่ได้บ้าง (และเมื่อใดที่คุณต้องการฟังก์ชันเพิ่มเติม)

ระบบนี้รองรับการจัดส่งได้สูงสุดประมาณ 10,000 รายการต่อนาที บนอินสแตนซ์ PostgreSQL เดียวที่มีเวิร์กช็อปเกอร์ 2-3 ตัว หากต้องการมากกว่านั้น จำเป็นต้องมีการเปลี่ยนแปลงสามอย่าง

ขั้นแรก ให้เปลี่ยนคิว PostgreSQL เป็น Redis Streams หรือ RabbitMQ SELECT FOR UPDATE SKIP LOCKED รูปแบบดังกล่าวทำให้เกิดการแย่งชิงการเขียนข้อมูลบนตารางคิวเมื่อมีปริมาณงานสูง การใช้ตัวกลางรับส่งข้อความโดยเฉพาะจะช่วยขจัดปัญหานี้ได้

ประการที่สอง เพิ่มการจำกัดอัตราการส่งคำขอต่อปลายทาง ปลายทางรับบางแห่งมีการจำกัดอัตราการส่งคำขอ (100 คำขอต่อนาที, 1,000 คำขอต่อชั่วโมง) หากไม่มีการจำกัดอัตราการส่งคำขอฝั่งไคลเอ็นต์ คุณจะใช้โควต้าเกินและถูกปฏิเสธ 429 ควรใช้โทเค็นบัคเก็ตต่อปลายทาง

ประการที่สาม เพิ่มการลงนามคำขอ การลงนาม HMAC-SHA256 บนข้อมูลจะช่วยให้ปลายทางผู้รับตรวจสอบได้ว่าเว็บฮุคมาจากระบบของคุณและไม่ได้ถูกแก้ไขระหว่างการส่ง นี่เป็นสิ่งจำเป็นพื้นฐานสำหรับระบบเว็บฮุคใดๆ ที่ส่งข้อมูลทางการเงิน

ระบบในบทความนี้เป็นพื้นฐาน มันจัดการกับปัญหาที่ซับซ้อน เช่น ตรรกะการลองใหม่ การตัดวงจร การวัดความหน่วง การวิเคราะห์ข้อความที่ส่งไม่ถึงปลายทาง ซึ่งเป็นสิ่งที่ระบบส่งเว็บฮุคทุกระบบต้องการ ไม่ว่าจะมีขนาดใหญ่แค่ไหนก็ตาม ส่วนประกอบเฉพาะที่คุณเพิ่มเข้าไป (ตัวกลางส่งข้อความ ตัวจำกัดอัตรา การลงนามคำขอ) ขึ้นอยู่กับปริมาณงานและข้อกำหนดด้านความปลอดภัยของคุณ

ส่วนที่สำคัญที่สุดกลับเป็นส่วนที่ทีมส่วนใหญ่มองข้ามไป นั่นคือ การวัดประสิทธิภาพของระบบการส่งมอบข้อมูลเอง หากคุณไม่สามารถตอบคำถามได้ว่า “ความล่าช้าในการส่งมอบข้อมูล P95 ไปยังปลายทาง X ในช่วง 24 ชั่วโมงที่ผ่านมาคือเท่าไร” แสดงว่าคุณกำลังทำงานโดยไม่รู้ข้อมูลอะไรเลย สร้างระบบการวัดผลก่อน แล้วทุกอย่างจะตามมาเอง

คำถามที่พบบ่อยเกี่ยวกับการส่งมอบ Webhook

ผู้ส่ง webhook ควรรับประกันการส่งเพียงครั้งเดียวหรือไม่?

โดยปกติแล้วไม่ใช่ ผู้ส่งควรทำให้การส่งซ้ำมองเห็นได้และระบุรหัสเหตุการณ์ที่เสถียร ในขณะที่ผู้รับควรทำให้การประมวลผลเป็นแบบไม่เปลี่ยนแปลงผลลัพธ์เดิม เพื่อให้สามารถส่งเหตุการณ์ได้มากกว่าหนึ่งครั้งโดยไม่ทำให้ผลลัพธ์ทางธุรกิจซ้ำซ้อน

ควรลองใหม่ในกรณีที่ล้มเหลวครั้งใดบ้าง?

ลองใหม่เฉพาะความล้มเหลวที่สัญญาของคุณจัดประเภทว่าเป็นความล้มเหลวชั่วคราว เช่น ข้อผิดพลาดเครือข่าย การหมดเวลา และการตอบสนองของเซิร์ฟเวอร์ที่เลือกไว้ อย่าลองใหม่ซ้ำๆ กับคำขอที่ไม่ถูกต้อง ความล้มเหลวในการตรวจสอบสิทธิ์ หรือข้อผิดพลาดถาวรอื่นๆ โดยไม่มีแนวทางการแก้ไขที่ชัดเจน

ทีมควรเปลี่ยนจากการใช้คิวที่เก็บข้อมูลในฐานข้อมูลเมื่อใด?

ควรปรับเปลี่ยนเมื่อการวัดปริมาณการแย่งใช้ทรัพยากร อายุของงานค้าง ปริมาณงาน หรือความต้องการในการกู้คืนระบบแสดงให้เห็นว่าคิวฐานข้อมูลไม่เป็นไปตามข้อตกลงการส่งมอบอีกต่อไป การตัดสินใจเรื่องความจุควรพิจารณาจากปริมาณงานที่สังเกตได้ ไม่ใช่จากการอ้างอิงอัตราการร้องขอโดยทั่วไป