Idempotency: كيف تجعل العملية آمنة عند تكرارها

كل نظام يرسل رسائل بين خدماته سوف يستقبل يوماً ما الرسالة نفسها مرتين، وهذا ليس خطأ في الطابور الذي تستخدمه، وإنما هو الثمن الذي تدفعه مقابل ألا تضيع الرسائل. وقد شرحنا في مقال الأنظمة الموزعة لماذا تختار أغلب الطوابير ضمان «مرة على الأقل At-Least-Once»، ورأينا في مفاهيم الطوابير كيف تعود الرسالة إلى ال Worker إذا توقف قبل التأكيد، وشرحنا مفتاح منع التكرار Idempotency Key في الطلبات القادمة من الكلاينت في دليل تصميم ال API، وكيف يحفظ داخل المعاملة نفسها في المحفظة الرقمية. فلن نكرر شيئاً من ذلك هنا، وإنما نركز على الطرف الذي لم نكتب له كوداً كاملاً بعد: المستهلك Consumer الذي يستقبل الرسائل من الطابور، وكيف تجعله يطبق كل رسالة مرة واحدة فقط مهما تكررت.

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

وسوف نناقش في هذا المقال ما يلي:

• ما معنى Idempotency بالضبط، ومن أين تأتي الرسالة المكررة في الواقع.
• العمليات الآمنة عند التكرار بطبيعتها، وكيف تعيد كتابة عملية غير آمنة لتصبح آمنة دون أي جدول إضافي.
• المستهلك الآمن عند التكرار Idempotent Consumer: جدول الرسائل المعالجة، وأي معرف تحفظ، ولماذا مكانه نفس المعاملة.
• الآثار خارج قاعدة البيانات مثل البريد واستدعاء واجهة خارجية، والرسائل التي تصل بغير ترتيبها، ومتى تحذف سجلات منع التكرار.
• تجربة عملية على Valkey Streams وPostgreSQL: مستهلك بسيط يضيف المبلغ مرتين، ثم مستهلك صحيح، ثم نكسر الحل بوضع علامة منع التكرار في Valkey.

ما معنى Idempotency؟

العملية الآمنة عند التكرار Idempotent هي العملية التي إذا نفذتها مرتين أو عشر مرات كانت النتيجة كما لو نفذتها مرة واحدة، والمثال الذي تعرفه هو زر المصعد، فالضغط عليه خمس مرات لا يحضر خمسة مصاعد، بينما الضغط على زر «اشتر الآن» خمس مرات في متجر سيئ التصميم قد ينشئ خمسة طلبات. وتعرف المواصفة RFC 9110 في القسم 9.2.2 هذه الخاصية لطرق HTTP، فتذكر أن PUT وDELETE والطرق الآمنة مثل GET متكررة الأثر Idempotent، أما POST فليست كذلك، ولهذا احتاجت الواجهات مفتاح منع التكرار للطلبات من نوع POST.

ولاحظ أن التعريف عن الأثر وليس عن الرد، فحذف طلب موجود يرد «تم الحذف»، وحذفه مرة ثانية قد يرد «غير موجود»، ولكن الأثر واحد: الطلب لم يعد موجوداً.

من أين تأتي الرسالة المكررة؟

الصورة التالية تبين الطريقين اللذين تأتي منهما الرسالة المكررة، وهما بالضبط ما يحدث في التجربة العملية:

• من طرف المستهلك: المستهلك يطبق الرسالة ويحفظ الأثر، ثم يتوقف قبل أن يرسل التأكيد Ack، فلا يعرف الوسيط أن الرسالة عولجت، فيعطيها لمستهلك آخر. وهذا لا يحتاج عطلاً كبيراً، فنشر نسخة جديدة يوقف المستهلكين في منتصف عملهم كل يوم.
• من طرف المنتج: المنتج يرسل الرسالة ولا يصله التأكيد بسبب انتهاء المهلة، فلا يعرف هل وصلت أم لا، فيرسلها مرة ثانية، وهذه المرة تصل نسختان، ولكل نسخة معرف مختلف في الوسيط.
• ومن خارج الطابور أيضاً: المهمة المجدولة التي تعمل مرتين لأن سيرفرين ظنا أن كلاً منهما المسؤول، والمستخدم الذي يضغط الزر مرتين، و Webhook يرسله مزود الدفع مرة ثانية لأن ردك تأخر.

وتقول AWS صراحة في توثيق الطوابير القياسية في SQS إن الرسالة قد تصل أكثر من مرة، وتطلب منك أن تصمم تطبيقك ليكون آمناً عند التكرار، أي ألا يتأثر سلباً عندما يعالج الرسالة نفسها أكثر من مرة. والخلاصة العملية: لا تحاول منع التكرار في الطريق، فهذا غير ممكن، وإنما اجعل الطرف الأخير يتحمله.

عمليات آمنة بطبيعتها، وأخرى تحتاج مفتاحاً

قبل أن تضيف جدولاً لمنع التكرار، اسأل: هل أستطيع أن أكتب العملية نفسها بطريقة لا يضرها التكرار؟ وغالباً تستطيع، والصورة التالية تبين الفرق:

العملية التي تضع قيمة نهائية، مثل «اجعل حالة الطلب shipped»، آمنة بطبيعتها، وتصبح أكثر أماناً إذا اشترطت الحالة السابقة، مثل UPDATE orders SET status = 'shipped' WHERE id = 1 AND status = 'paid'، فالمرة الثانية لا تغير شيئاً وتخبرك بذلك، لأن عدد الصفوف المتغيرة صفر، فتعرف أنها مكررة. وهذا هو مبدأ آلة الحالات State Machine، فكل حدث ينقل الكيان من حالة محددة إلى حالة محددة، والحدث الذي يأتي والكيان ليس في الحالة المتوقعة إما مكرر أو متأخر، فتسجله وتتجاهله. أما العملية التي تضيف أو تنقص، مثل balance = balance + 50 أو qty = qty - 1، فكل تكرار لها أثر جديد، وقد رأينا في التجربة أن المرة الثانية أنقصت المخزون من 9 إلى 8.

وأحياناً تستطيع أن تحول العملية الثانية إلى الأولى، فبدل أن ترسل حدث «أضف 50 إلى الرصيد» ترسل حدث «الدفعة رقم 7781 بقيمة 50»، ويضيفها المستهلك كصف في جدول الدفعات بمفتاح أساسي هو رقم الدفعة، ثم يحسب الرصيد من الجدول، فيصبح التكرار محاولة إدخال صف موجود يرفضها القيد، وهذه هي فكرة دفتر القيود في المحفظة الرقمية نفسها. وإذا لم تستطع ذلك فأنت تحتاج مستهلكاً آمناً عند التكرار، وهو موضوع القسم التالي.

المستهلك الآمن عند التكرار Idempotent Consumer

الفكرة التي يصفها نمط Idempotent Consumer بسيطة: المستهلك يحفظ معرف كل رسالة عالجها، فإذا جاءته رسالة بمعرف موجود تجاهلها. ولكن التفاصيل هي التي تفرق بين حل يعمل وحل يبدو أنه يعمل، وهي ثلاثة:

أي معرف تحفظ؟

الوسيط يعطي كل رسالة معرفاً، مثل 1791609321291-0 في Valkey Streams أو رقم الإزاحة Offset في Kafka، والحل البسيط هو أن تحفظ هذا المعرف. وهذا يحمي من الطريق الأول فقط، أي إعادة نفس الرسالة بعد توقف المستهلك، ولكنه لا يحمي من الطريق الثاني، لأن المنتج الذي يعيد الإرسال ينشئ رسالة جديدة بمعرف جديد في الوسيط، وقد رأيت ذلك في التجربة: الحدث 2 وصل بالمعرفين 1791609321290-0 و1791609321292-0. لذلك يجب أن يحمل الحدث نفسه معرفاً يولده المنتج مرة واحدة قبل أول محاولة إرسال، مثل UUID أو ULID كما في دليل المعرفات الفريدة، ويبقى ثابتاً في كل المحاولات، وهذا هو المعرف الذي تحفظه.

أين تحفظ العلامة؟

العلامة «عولجت هذه الرسالة» يجب أن تحفظ في نفس المعاملة Transaction التي تحفظ الأثر، والصورة التالية تبين لماذا:

إذا كانت العلامة في نظام آخر، مثل Valkey، فهناك لحظة بين حفظ العلامة وحفظ الأثر، فإذا توقف المستهلك فيها بقيت العلامة دون أثر، فتأتي الرسالة المعادة وترى العلامة فتتجاهلها، وتضيع الدفعة. وإذا عكست الترتيب فحفظت الأثر ثم العلامة، فالتوقف بينهما يعيدك إلى المشكلة الأصلية: أثر مرتين. والحل الوحيد الذي لا توجد فيه هذه اللحظة هو أن تكون العلامة صفاً في قاعدة البيانات نفسها، بقيد فريد، داخل نفس المعاملة، فيستخدم INSERT … ON CONFLICT DO NOTHING، ويذكر توثيق INSERT في PostgreSQL أن هذه الصيغة تضمن نتيجة الإدخال أو عدمه حتى مع التزامن، فإذا وصلت النسختان إلى مستهلكين مختلفين في نفس اللحظة انتظر الثاني حتى تنتهي معاملة الأول، ثم لا يدخل شيئاً.

متى ترسل التأكيد؟

بعد COMMIT، وليس قبله. فإذا أكدت قبل الحفظ ثم توقفت فقد ضاعت الرسالة، وإذا أكدت بعده ثم توقفت قبل التأكيد فستعود الرسالة، والجدول يتجاهلها، وهذا هو المطلوب بالضبط. ولاحظ أن الرسالة المكررة يجب أن تؤكد أيضاً بعد تجاهلها، وإلا عادت مرة بعد مرة، وانتهى بها الأمر في Dead Letter Queue وهي لم تفشل أصلاً.

الآثار خارج قاعدة البيانات

كل ما سبق يعمل لأن الأثر صف في قاعدة البيانات نفسها، ولكن ماذا إذا كان الأثر إرسال بريد، أو استدعاء بوابة دفع، أو رسالة SMS؟ لا توجد معاملة واحدة تجمع قاعدة بياناتك ومزود البريد، ولذلك أمامك طريقان:

• إذا كان الطرف الآخر يقبل مفتاح منع التكرار، كما تفعل بوابات الدفع الكبيرة، فأرسل له معرف الحدث نفسه مفتاحاً، فإذا أرسلت الطلب مرتين نفذه مرة واحدة، وتكون المسؤولية عنده.
• وإذا لم يقبله، فاجعل الإرسال نفسه حدثاً في جدول داخل قاعدتك، أي في جدول Outbox بقيد فريد على معرف الحدث، ثم يرسله عامل منفصل ويسجل أنه أرسله. وهنا تبقى لحظة صغيرة بين الإرسال والتسجيل قد تنتج رسالة مكررة نادراً، وهذا هو الحد الذي لا يمكن تجاوزه مع طرف لا يتعاون، وهو نفس الحد الذي ذكرناه عن «مرة واحدة بالضبط» مع الأنظمة الخارجية في مقال الأنظمة الموزعة، فقرر مع فريق العمل: هل بريد مكرر نادر مقبول، أم بريد ضائع نادر؟ وغالباً المكرر أهون.

رسائل تصل بغير ترتيبها

منع التكرار لا يحل مشكلة قريبة منها: الرسالة القديمة التي تصل بعد الجديدة. فإذا وصل حدث «عنوان العميل أصبح الرياض» ثم وصل بعده حدث أقدم «عنوان العميل أصبح جدة» بسبب إعادة المحاولة، فكلاهما جديد في جدول منع التكرار، والنتيجة عنوان خاطئ. والحل أن يحمل كل حدث رقم إصدار Version للكيان، يزيده المنتج مع كل تغيير، والمستهلك يطبق التغيير بشرط، مثل UPDATE customers SET city = $1, version = $2 WHERE id = $3 AND version < $2، فالحدث الأقدم لا يغير شيئاً. وهذا هو نفس مبدأ القفل المتفائل Optimistic Locking الذي رأيناه في نظام حجز التذاكر.

متى تحذف سجلات منع التكرار؟

جدول الرسائل المعالجة يكبر مع كل رسالة، فإذا كان عندك مليون رسالة في اليوم فهو 365 مليون صف في السنة، ولذلك تحذف القديم بمهمة دورية. والسؤال: كم تحتفظ؟ والإجابة: أطول من أطول مدة يمكن أن تتأخر فيها نسخة مكررة. فإذا كان الوسيط يحتفظ بالرسائل سبعة أيام، والمنتج يعيد المحاولة لمدة يوم، فاحتفظ بالسجلات أكثر من سبعة أيام بهامش. وإذا كبر الجدول جداً فقسمه بالتاريخ Partitioning، فيصبح الحذف حذف جزء كامل بدل ملايين الصفوف، كما شرحنا في دليل الجداول الضخمة في PostgreSQL.

تجربة عملية: مستهلك يستقبل الرسالة مرتين

في هذه التجربة نستخدم Valkey Streams وسيطاً، لأنه يعطيك سلوك «مرة على الأقل» الحقيقي بأوامر قليلة: المستهلك يقرأ الرسائل بالأمر XREADGROUP ضمن مجموعة مستهلكين Consumer Group، فتبقى كل رسالة في قائمة الرسائل المعلقة Pending Entries List حتى يؤكدها بالأمر XACK، وإذا لم يؤكدها يستطيع مستهلك آخر أن يأخذها بالأمر XAUTOCLAIM بعد أن تبقى معلقة مدة محددة، كما يشرح توثيق Streams في Valkey وتوثيق XAUTOCLAIM. والسيناريو: المنتج يرسل خمس دفعات من 50 ريالاً إلى محفظة، ويعيد إرسال الدفعة الثانية بعد انتهاء مهلة، والمستهلك c1 يتوقف فجأة أثناء معالجة الدفعة الثالثة، ثم يبدأ المستهلك c2 ويكمل. والرصيد الصحيح في النهاية 250.00.

المتطلبات: جهاز عليه Docker Engine وDocker Compose، ولا تحتاج نشر أي منفذ على جهازك.

ملفات التجربة

أنشئ مجلداً فيه أربعة ملفات، الأول compose.yaml:

name: lab-idem
services:
  db:
    image: postgres:18.6
    container_name: lab-idem-db
    environment:
      POSTGRES_PASSWORD: lab-only-password
      POSTGRES_DB: wallet
    volumes:
      - ./schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d wallet"]
      interval: 2s
      retries: 30
  broker:
    image: valkey/valkey:9.1.2-alpine
    container_name: lab-idem-valkey
    command: ["valkey-server", "--appendonly", "yes"]
    healthcheck:
      test: ["CMD", "valkey-cli", "ping"]
      interval: 2s
      retries: 30
  app:
    build: .
    image: lab-idem-app:1
    container_name: lab-idem-app
    environment:
      DATABASE_URL: postgresql://postgres:lab-only-password@db:5432/wallet
      VALKEY_URL: valkey://broker:6379/0
    depends_on:
      db:
        condition: service_healthy
      broker:
        condition: service_healthy
    profiles: ["run"]

لاحظ أن Valkey هنا يعمل بالخيار --appendonly yes، لأنه هنا وسيط رسائل وليس Cache، فيجب ألا تضيع الرسائل إذا أعيد تشغيله.

والملف الثاني schema.sql، وفيه جدول المحفظة وجدول الرسائل المعالجة:

CREATE TABLE wallets (
    id              text   PRIMARY KEY,
    balance_halala  bigint NOT NULL
);
INSERT INTO wallets VALUES ('w1', 0);

-- one row per message a consumer has already applied; the key is the business event id, not the broker's id
CREATE TABLE processed_messages (
    consumer      text        NOT NULL,
    event_id      uuid        NOT NULL,
    processed_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer, event_id)
);

CREATE TABLE orders (
    id      int  PRIMARY KEY,
    status  text NOT NULL
);
CREATE TABLE stock (
    sku  text PRIMARY KEY,
    qty  int  NOT NULL
);

والملف الثالث Dockerfile:

FROM python:3.13.16-slim
RUN pip install --no-cache-dir "psycopg[binary]==3.3.6" "valkey==6.1.1"
WORKDIR /app
COPY consumer.py .
ENTRYPOINT ["python", "-u", "consumer.py"]

والملف الرابع consumer.py:

# A wallet consumer on a Valkey stream (at-least-once), first written the simple way, then made idempotent.
#   python consumer.py demo simple | demo dedup-table | demo dedup-valkey | natural
import os, subprocess, sys, time, uuid

import psycopg
import valkey

r = valkey.Valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True)
DB = os.environ["DATABASE_URL"]
STREAM, GROUP = "payments", "wallet"
NAME = os.environ.get("NAME", "c1")

def sar(halala):
    return f"{halala / 100:.2f}"

def publish():
    events = [str(uuid.uuid4()) for _ in range(5)]
    for n, ev in enumerate(events, 1):
        sid = r.xadd(STREAM, {"n": n, "event_id": ev, "wallet": "w1", "amount": 5000})
        print(f"producer: {sid}  event {n} ({ev[:8]}) +50.00")
    # the producer's call for event 2 timed out, so it sent the same event again: a new stream id, the same event id
    sid = r.xadd(STREAM, {"n": 2, "event_id": events[1], "wallet": "w1", "amount": 5000})
    print(f"producer: {sid}  event 2 ({events[1][:8]}) +50.00   <- sent again after a timeout")

# ---------- three ways to apply one message ----------
def apply_simple(conn, msg):
    with conn.transaction():
        bal = conn.execute("UPDATE wallets SET balance_halala = balance_halala + %s WHERE id = %s RETURNING balance_halala",
                           (int(msg["amount"]), msg["wallet"])).fetchone()[0]
    return f"applied, balance {sar(bal)}"

def apply_dedup_table(conn, msg):
    with conn.transaction():          # the mark and the effect are committed together, or not at all
        first = conn.execute("INSERT INTO processed_messages (consumer, event_id) VALUES ('wallet', %s) ON CONFLICT DO NOTHING",
                             (msg["event_id"],)).rowcount
        if not first:
            return "duplicate, skipped"
        bal = conn.execute("UPDATE wallets SET balance_halala = balance_halala + %s WHERE id = %s RETURNING balance_halala",
                           (int(msg["amount"]), msg["wallet"])).fetchone()[0]
    return f"applied, balance {sar(bal)}"

def apply_dedup_valkey(conn, msg, crash=False):
    if not r.set(f"seen:{msg['event_id']}", 1, nx=True, ex=86400):   # the mark lives in another system
        return "duplicate, skipped"
    if crash:
        print(f"  {NAME}: marked event {msg['n']} as seen in Valkey, then crashed before the database commit")
        os._exit(1)
    return apply_simple(conn, msg)

def consume(variant, crash_on):
    """crash_on: the event number whose handling ends in a crash (0 = none)"""
    with psycopg.connect(DB, autocommit=True) as conn:
        # 1) messages that another consumer took and never acknowledged come back once they are idle long enough
        _, claimed, _ = r.xautoclaim(STREAM, GROUP, NAME, min_idle_time=1000, start_id="0-0")
        batch = [(sid, f, "redelivered") for sid, f in claimed]
        # 2) then the new messages
        for _, msgs in r.xreadgroup(GROUP, NAME, {STREAM: ">"}, count=100, block=1000) or []:
            batch += [(sid, f, "new") for sid, f in msgs]
        for sid, msg, kind in batch:
            crash = crash_on and msg["n"] == str(crash_on)
            if variant == "dedup-valkey":
                result = apply_dedup_valkey(conn, msg, crash)
            else:
                result = (apply_simple if variant == "simple" else apply_dedup_table)(conn, msg)
            print(f"  {NAME}: {sid}  event {msg['n']} ({kind}) -> {result}")
            if crash:
                print(f"  {NAME}: crashed before XACK")
                os._exit(1)
            r.xack(STREAM, GROUP, sid)

def reset():
    r.delete(STREAM, *r.keys("seen:*"))
    r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
    with psycopg.connect(DB, autocommit=True) as c:
        c.execute("UPDATE wallets SET balance_halala = 0")
        c.execute("TRUNCATE processed_messages")

def demo(variant):
    reset()
    publish()
    print(f"\nconsumer c1 ({variant}) crashes while handling event 3:")
    env = dict(os.environ, NAME="c1")
    subprocess.run([sys.executable, "-u", __file__, "consume", variant, "3"], env=env)
    time.sleep(1.2)                                    # the unacknowledged message must be idle for 1 s
    print(f"\nconsumer c2 ({variant}) starts:")
    subprocess.run([sys.executable, "-u", __file__, "consume", variant, "0"], env=dict(os.environ, NAME="c2"))
    with psycopg.connect(DB) as c:
        bal = c.execute("SELECT balance_halala FROM wallets WHERE id = 'w1'").fetchone()[0]
        marks = c.execute("SELECT count(*) FROM processed_messages").fetchone()[0]
    print(f"\nfive different payments of 50.00 were sent. wallet balance: {sar(bal)}   (expected 250.00)")
    print(f"unacknowledged messages: {r.xpending(STREAM, GROUP)['pending']}   rows in processed_messages: {marks}")

def natural():
    with psycopg.connect(DB, autocommit=True) as c:
        c.execute("DELETE FROM orders"); c.execute("DELETE FROM stock")
        c.execute("INSERT INTO orders VALUES (1, 'paid')"); c.execute("INSERT INTO stock VALUES ('A1', 10)")
        for i in (1, 2):
            n = c.execute("UPDATE orders SET status = 'shipped' WHERE id = 1 AND status = 'paid'").rowcount
            q = c.execute("UPDATE stock SET qty = qty - 1 WHERE sku = 'A1' RETURNING qty").fetchone()[0]
            print(f"delivery {i}: set order 1 shipped -> {n} row(s) changed   |   stock A1 minus 1 -> qty {q}")

if __name__ == "__main__":
    cmd = sys.argv[1]
    if cmd == "demo": demo(sys.argv[2])
    elif cmd == "consume": consume(sys.argv[2], int(sys.argv[3]))
    elif cmd == "natural": natural()

في الكود أعلاه لاحظ التالي:

• كل حدث يحمل event_id يولده المنتج مرة واحدة، والإرسال المكرر للحدث 2 يستخدم نفس المعرف.
• الدالة apply_simple تضيف المبلغ فقط، والدالة apply_dedup_table تدخل العلامة وتضيف المبلغ في معاملة واحدة، والدالة apply_dedup_valkey تضع العلامة في Valkey بالأمر SET NX قبل أن تلمس قاعدة البيانات.
• التوقف يحدث بالأمر os._exit(1)، وهو يقتل العملية فوراً دون أي تنظيف، كما يحدث عند انقطاع الكهرباء أو قتل ال Container.
• المستهلك c2 يبدأ بالأمر XAUTOCLAIM بمدة 1000 ملي ثانية، فيأخذ كل ما تركه c1 دون تأكيد.

الآن نشغل قاعدة البيانات وValkey ونبني صورة المستهلك:

docker compose up -d --wait
docker compose --profile run build app

المستهلك البسيط

docker compose run --rm -q app demo simple

والمخرج سوف يكون كما يلي:

producer: 1791609321289-0  event 1 (f443727e) +50.00
producer: 1791609321290-0  event 2 (b215c4b0) +50.00
producer: 1791609321291-0  event 3 (2e0e832a) +50.00
producer: 1791609321291-1  event 4 (d83440f0) +50.00
producer: 1791609321291-2  event 5 (0f54ee00) +50.00
producer: 1791609321292-0  event 2 (b215c4b0) +50.00   <- sent again after a timeout

consumer c1 (simple) crashes while handling event 3:
  c1: 1791609321289-0  event 1 (new) -> applied, balance 50.00
  c1: 1791609321290-0  event 2 (new) -> applied, balance 100.00
  c1: 1791609321291-0  event 3 (new) -> applied, balance 150.00
  c1: crashed before XACK

consumer c2 (simple) starts:
  c2: 1791609321291-0  event 3 (redelivered) -> applied, balance 200.00
  c2: 1791609321291-1  event 4 (redelivered) -> applied, balance 250.00
  c2: 1791609321291-2  event 5 (redelivered) -> applied, balance 300.00
  c2: 1791609321292-0  event 2 (redelivered) -> applied, balance 350.00

five different payments of 50.00 were sent. wallet balance: 350.00   (expected 250.00)
unacknowledged messages: 0   rows in processed_messages: 0

في المخرج أعلاه لاحظ التالي:

• الحدث 3 طبق مرتين: مرة عند c1 قبل أن يتوقف، ومرة عند c2 بعد أن أعاده الوسيط، والحدث 2 طبق مرتين لأن المنتج أرسله مرتين.
• الأحداث 4 و5 ظهرت عند c2 بكلمة redelivered، والسبب أن c1 قرأ كل الرسائل الست في قراءة واحدة بالخيار count=100، فأصبحت كلها معلقة باسمه، وهذا سلوك عادي في أي مستهلك يقرأ دفعات.
• كل الرسائل أكدت في النهاية، فالوسيط يرى أن كل شيء تم بنجاح، والخطأ لا يظهر إلا في الرصيد.

المستهلك مع جدول الرسائل المعالجة

docker compose run --rm -q app demo dedup-table

والمخرج سوف يكون كما يلي:

producer: 1791609333098-0  event 1 (708d22ff) +50.00
producer: 1791609333099-0  event 2 (f1a7cf24) +50.00
producer: 1791609333099-1  event 3 (334d2484) +50.00
producer: 1791609333100-0  event 4 (6a73fdad) +50.00
producer: 1791609333100-1  event 5 (de643563) +50.00
producer: 1791609333101-0  event 2 (f1a7cf24) +50.00   <- sent again after a timeout

consumer c1 (dedup-table) crashes while handling event 3:
  c1: 1791609333098-0  event 1 (new) -> applied, balance 50.00
  c1: 1791609333099-0  event 2 (new) -> applied, balance 100.00
  c1: 1791609333099-1  event 3 (new) -> applied, balance 150.00
  c1: crashed before XACK

consumer c2 (dedup-table) starts:
  c2: 1791609333099-1  event 3 (redelivered) -> duplicate, skipped
  c2: 1791609333100-0  event 4 (redelivered) -> applied, balance 200.00
  c2: 1791609333100-1  event 5 (redelivered) -> applied, balance 250.00
  c2: 1791609333101-0  event 2 (redelivered) -> duplicate, skipped

five different payments of 50.00 were sent. wallet balance: 250.00   (expected 250.00)
unacknowledged messages: 0   rows in processed_messages: 5

نفس التوقف في نفس المكان، ولكن الحدث 3 وجد علامته في الجدول فتجاهله c2، والحدث 2 المرسل مرة ثانية وجد علامته أيضاً رغم أن معرفه في الوسيط مختلف، لأن العلامة هي event_id وليست معرف الوسيط، فأصبح الرصيد 250.00 وفي الجدول خمسة صفوف لخمس دفعات.

نكسر الحل: العلامة في Valkey

كثير من الفرق تضع علامة منع التكرار في Valkey لأنه أسرع ولأن مدة الصلاحية تحذف العلامات القديمة وحدها، فلنر ماذا يحدث إذا توقف المستهلك بعد وضع العلامة وقبل حفظ الأثر:

docker compose run --rm -q app demo dedup-valkey

والمخرج سوف يكون كما يلي:

producer: 1791609349120-0  event 1 (c8530ce2) +50.00
producer: 1791609349121-0  event 2 (f7518961) +50.00
producer: 1791609349122-0  event 3 (c9fe37dc) +50.00
producer: 1791609349122-1  event 4 (63d210d7) +50.00
producer: 1791609349122-2  event 5 (e10ff360) +50.00
producer: 1791609349123-0  event 2 (f7518961) +50.00   <- sent again after a timeout

consumer c1 (dedup-valkey) crashes while handling event 3:
  c1: 1791609349120-0  event 1 (new) -> applied, balance 50.00
  c1: 1791609349121-0  event 2 (new) -> applied, balance 100.00
  c1: marked event 3 as seen in Valkey, then crashed before the database commit

consumer c2 (dedup-valkey) starts:
  c2: 1791609349122-0  event 3 (redelivered) -> duplicate, skipped
  c2: 1791609349122-1  event 4 (redelivered) -> applied, balance 150.00
  c2: 1791609349122-2  event 5 (redelivered) -> applied, balance 200.00
  c2: 1791609349123-0  event 2 (redelivered) -> duplicate, skipped

five different payments of 50.00 were sent. wallet balance: 200.00   (expected 250.00)
unacknowledged messages: 0   rows in processed_messages: 0

الحل منع التكرار فعلاً، فالحدث 2 المكرر تجوهل، ولكنه أضاع الدفعة الثالثة، لأن c2 وجد علامتها في Valkey فظن أنها طبقت، والرصيد 200.00 بدل 250.00. وهذا الخطأ أسوأ من التكرار، لأن التكرار يظهر في شكوى العميل الذي دفع مرتين، أما الضياع فقد لا يكتشفه أحد إلا في المطابقة الشهرية. ولا يعني هذا أن Valkey لا مكان له، فتستطيع أن تضعه أمام الجدول كفحص سريع يتجاهل أغلب المكرر قبل أن يفتح معاملة، بشرط أن يبقى الجدول داخل المعاملة هو الحكم الأخير.

العملية الآمنة بطبيعتها

وأخيراً نطبق نفس الحدث مرتين على عمليتين، الأولى تضع حالة بشرط والثانية تنقص المخزون:

docker compose run --rm -q app natural

والمخرج سوف يكون كما يلي:

delivery 1: set order 1 shipped -> 1 row(s) changed   |   stock A1 minus 1 -> qty 9
delivery 2: set order 1 shipped -> 0 row(s) changed   |   stock A1 minus 1 -> qty 8

التحديث المشروط لم يغير شيئاً في المرة الثانية، وأخبرنا بذلك بعدد الصفوف، أما إنقاص المخزون فطبق مرتين.

حذف التجربة

docker compose --profile run down -v
docker image rm lab-idem-app:1

أخطاء شائعة

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

الخلاصة

وصلنا لنهاية الموضوع، وأهم ما فيه:

• الرسالة المكررة نتيجة طبيعية لضمان «مرة على الأقل»، وتأتي من مستهلك توقف قبل التأكيد ومن منتج أعاد الإرسال، فاجعل الطرف الأخير يتحملها.
• العملية التي تضع قيمة نهائية بشرط آمنة بطبيعتها، والعملية التي تضيف أو تنقص تحتاج مفتاحاً.
• المفتاح هو معرف الحدث الذي يولده المنتج مرة واحدة، وليس معرف الوسيط.
• العلامة تحفظ بقيد فريد في نفس المعاملة التي تحفظ الأثر، والتأكيد يرسل بعد COMMIT، وفي التجربة أعطى ذلك 250.00 بدل 350.00.
• العلامة في نظام آخر منعت التكرار ولكنها أضاعت دفعة كاملة، فالرصيد 200.00.
• الآثار الخارجية تحتاج مفتاحاً يقبله الطرف الآخر أو جدول Outbox، والأحداث غير المرتبة تحتاج رقم إصدار.

وفي المقال التالي من هذا القسم نستخدم كل ذلك في عملية تمتد على ثلاث خدمات، حيث كل خدمة مستهلك آمن عند التكرار:

دليلالمعاملات الموزعة: Saga ونمط Outbox بدل 2PCكيف تنفذ عملية تمتد على عدة خدمات دون 2PC، بخطوات محلية وتعويضات وجدول Outbox لا يضيع فيه أي حدث.دليلكيف تصمم API لا تضطر إلى كسره بعد سنةمبادئ تصميم API يتطور دون أن يكسر التطبيقات التي تعتمد عليه، وكيف تختار الشكل المناسب لنظامك.دليلالرقم المتسلسل أم UUID؟ كيف تختار المعرف الفريدلماذا يفشل الرقم المتسلسل في الروابط وعند تعدد السيرفرات، وكيف يؤثر نوع المعرف الذي تختاره على فهرس قاعدة البيانات.

سجل التحديثات

• أكتوبر 2026: كتابة المقال، وتنفيذ التجربة العملية على Docker Engine 29.8.0 وDocker Compose 5.5.1 مع PostgreSQL 18.6 وValkey 9.1.2 وPython 3.13.16 وpsycopg 3.3.6 وvalkey-py 6.1.1.