بناء نظام توصيل Webhook: ردود إرسال موثوقة في عام 2026

آخر تحديث في 24 يونيو 2026 بواسطة سيزار فيكسون

الإجابة المباشرة: تحتاج خدمة توصيل روابط الويب الموثوقة أو خدمة توصيل الإحالات التابعة إلى قائمة انتظار متينة، ومحاولات إعادة محدودة مع تذبذب، وكسر الدائرة لكل نقطة نهاية، وحماية من التكرار، والتحقق من التوقيع، وقياس عن بعد يوضح ما إذا كان التسليم يتباطأ قبل أن يفشل.

يُبقي هذا التطبيق نموذج التنفيذ صغيرًا عمدًا: عمال بايثون وقاعدة بيانات PostgreSQL لإدارة قائمة الانتظار وسجل التسليم وحالة نقاط النهاية. وهو نقطة انطلاق عملية لبرمجيات SaaS الموجهة للشركات، وعمليات التسويق بالعمولة في مجال الألعاب الإلكترونية، وأي منتج يتطلب إرسال أحداث خارجية دون اعتبار أي مهلة HTTP خسارةً في التحويل.

ما يجب أن يضمنه عقد التسليم

تأتي إجراءات الأمان والتحكم في النسخ المكررة قبل ضبط إعادة المحاولة

تعامل مع محتوى الطلب الصادر كبيانات تشغيلية حساسة. وقّع محتوى الطلب بدقة، وأضف طابعًا زمنيًا للتسليم ومعرّف حدث ثابت، وقم بتغيير أسرار التوقيع، وتأكد من أن التطبيق المُستقبِل يتجاهل إعادة تشغيل الحدث نفسه بأمان. لا يُعدّ نجاح استجابة HTTP دليلًا على تطبيق حدث تجاري مرة واحدة فقط؛ بل يجب على الطرف المُستقبِل اتخاذ هذا القرار.

بالنسبة لسير عمل إعادة توجيه العملاء في التسويق بالعمولة أو ألعاب الإنترنت، تجنب تضمين معرّفات النقرات، ومعرّفات التحويل، والحقول المتعلقة بالدفع في سجلات النظام إلا إذا كانت ضوابط الوصول وقواعد الاحتفاظ بالبيانات واضحة. يُعدّ مرجع Scaleo المذكور أدناه مثالاً فقط على زمن استجابة إعادة توجيه العملاء في التسويق بالعمولة، ولا يُغني عن توثيق اتفاقية التسليم بين خدماتك.

مراجع مفيدة للتنفيذ: التحقق من توقيع الويب هوك من Stripe, PostgreSQL SELECT و SKIP LOCKEDو إرشادات 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)

لماذا يُفضّل استخدام التذبذب الكامل بدلاً من التذبذب غير المرتبط أو التذبذب المتساوي؟ نشرت AWS تحليلاً شاملاً حول هذا الموضوع. يُنتج التذبذب الكامل (الذي يُوزّع عشوائياً بين الصفر والحد الأقصى الأسي) أقل وقت إنجاز إجمالي لجميع العملاء. أما التذبذب المتساوي (الذي يُوزّع عشوائياً بين نصف الحد الأقصى والحد الأقصى الكامل) فهو أكثر تحفظاً ولكنه أبطأ في تفريغ قائمة إعادة المحاولات المتراكمة. بالنسبة لتسليم Webhook حيث يوجد العديد من نقاط النهاية المستقلة، يُعدّ التذبذب الكامل الخيار الأمثل لأن محاولات إعادة كل نقطة نهاية مستقلة - أي لا يوجد تنسيق بينها.

قاطع الدائرة: توقف عن تكرار الأخطاء في نقاط النهاية المكسورة

يمنع نمط قاطع الدائرة نظامك من إهدار الموارد على نقاط النهاية التي تفشل باستمرار. فبدونه، تتراكم على نقطة النهاية المعطلة مئات من محاولات إعادة الاتصال المعلقة التي تنتهي جميعها بمهلة 15 ثانية، مما يستنزف قدرة نظامك على عمليات تسليم لن تنجح أبدًا.

الثعبان

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

إنّ آلية التصعيد من حالة "نصف مفتوح" إلى حالة "مفتوح" مع مضاعفة فترة الانتظار هي التفصيل الذي تغفله معظم التطبيقات. إذا فشلت نقطة نهاية أثناء عملية التحقق (حالة "نصف مفتوح")، فلا يُنصح بإعادة المحاولة بعد 5 دقائق أخرى، لأن نقطة النهاية لا تزال معطلة. لذا، يُنصح بمضاعفة فترة الانتظار إلى 10 دقائق، ثم إلى 20 دقيقة، بحد أقصى ساعة واحدة. هذا يمنع قاطع الدائرة من أن يصبح آلية ضغط دورية.

تتبع النسبة المئوية لزمن الاستجابة

المتوسطات مضللة. قد يستجيب جهاز طرفي بمتوسط ​​زمن استجابة 200 مللي ثانية في 50 مللي ثانية في 95% من الحالات، وفي 3,000 مللي ثانية في النسبة المتبقية (5%). يبدو المتوسط ​​جيدًا ظاهريًا، لكن نسبة P95 تكشف عن مشكلة تؤثر على عملية تسليم واحدة من كل 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. بالنسبة لتسليم إشعارات الويب ضمن نافذة زمنية مدتها 24 ساعة، فإن حساب النسب المئوية الدقيقة على البيانات الأولية سريع بما يكفي حتى حوالي 100,000 قياس لكل نقطة نهاية.

استخدم get_slow_endpoints أقوم بتشغيل هذه الوظيفة كفحص دوري كل 15 دقيقة. يتم وضع علامة على نقاط النهاية التي تتجاوز نسبة P95 فيها ثانيتين للتحقيق. أما نقاط النهاية التي تتجاوز نسبة 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 ثوانٍ، على وشك أن تبدأ في تجاوز مهلة الاتصال.

لا يقتصر دور قائمة الرسائل غير القابلة للتسليم على التخزين فقط.

تُطبّق معظم الفرق نظام قائمة انتظار الرسائل غير القابلة للتسليم، حيث تُخزّن فيها طلبات الويب الفاشلة. ويتم فحصها بشكل دوري أثناء الاستجابة للحوادث. وهذا يُعدّ هدراً للموارد.

تُعدّ قائمة الرسائل غير القابلة للتسليم أهمّ مجموعة بيانات لديك لتصحيح الأخطاء. يُمثّل كلّ صفّ فيها عملية تسليم حاول نظامك إيصالها عدّة مرّات وفشل. يُخبرك نمط الرسائل غير القابلة للتسليم بأمور لا تُوفّرها مقاييس النجاح أبدًا.

الثعبان

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

عندما أراجع الرسائل غير القابلة للنشر، أبحث عن ثلاثة أنماط.

حالات فشل المجموعة: خمسون رسالة غير قابلة للاستجابة لنفس نقطة النهاية في نفس الساعة تعني أن نقطة النهاية تعطلت ولم تستعد للعمل خلال فترة إعادة المحاولة. الإجراء المطلوب: تمديد فترة إعادة المحاولة أو إعادة جدولة الطلبات يدويًا.

أنماط رموز الحالة: ارتفاع مفاجئ في رسائل الخطأ 401/403 يعني أن نقطة النهاية قد غيّرت بيانات الاعتماد ولم يقم أحد بتحديث إعدادات Webhook. أما ارتفاع مفاجئ في رسائل الخطأ 429 (عدد كبير جدًا من الطلبات) فيعني أنك تتجاوز الحد المسموح به لمعدل الطلبات، ويجب عليك تقييد معدل الطلبات.

التراكم التدريجي: من 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 يُعدّ معالج SIGTERM ضروريًا لإيقاف التشغيل بسلاسة في بيئات الحاويات. فعندما يُرسل 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

هل ينبغي لمرسل إشعار الويب أن يعد بتسليم الرسالة مرة واحدة فقط؟

عادةً لا. يجب على المرسل إظهار عمليات إعادة المحاولة وتوفير معرف حدث ثابت؛ ويجب على المتلقي جعل المعالجة متكررة بحيث يمكن تسليم الحدث أكثر من مرة دون تكرار نتيجة العمل.

ما هي حالات الفشل التي يجب إعادة محاولتها؟

أعد المحاولة فقط في حالات الفشل التي يصنفها عقدك على أنها عابرة، مثل أخطاء الشبكة، وانتهاء المهلة، واستجابات الخادم المحددة. لا تعيد محاولة الطلبات غير الصحيحة، أو حالات فشل المصادقة، أو غيرها من الأخطاء الدائمة بشكل متكرر دون وجود مسار واضح للمعالجة.

متى ينبغي للفريق أن يتجاوز نظام قائمة الانتظار المدعومة بقاعدة البيانات؟

يجب اتخاذ إجراء عند قياس مستوى التنافس، أو عمر تراكم الطلبات، أو معدل نقل البيانات، أو الحاجة إلى استعادة العمليات، حيث يُشير ذلك إلى أن قائمة انتظار قاعدة البيانات لم تعد تفي بمتطلبات عقد التسليم. ينبغي أن تستند قرارات السعة إلى حجم العمل المُلاحظ، وليس إلى ادعاء عام بشأن معدل الطلبات.