مستوى متوسط

لو بتعدّل صف في PostgreSQL وبتبعت event لـ Kafka في نفس الـ HTTP request، عندك مشكلة صامتة بتنفجر مرة واحدة كل أسبوعين. الـ DB بتتحدّث، والـ produce بيفشل لأن الشبكة وقعت 800ms، والـ user بيرجعله 200 OK. النتيجة: الرصيد اتخصم لكن الإيميل ما اتبعتش، نظام الـ analytics ما عرفش، والـ search index بقى متأخر. هنا هتفهم Outbox Pattern وازاي بيحل ده بترانزاكشن واحد على الـ DB بدل اتنين منفصلتين.

Outbox Pattern: حلّ مشكلة Dual Write بأبسط فكرة معمارية ممكنة

المشكلة باختصار: Dual Write Problem

السيناريو: تطبيق fintech بيعمل خصم من حساب وبيبعت event باسم balance_changed لـ Kafka، علشان نظام الإشعارات والـ analytics والـ fraud detection كلهم يعرفوا.

# الكود الهش (الطريقة الشائعة الغلط)
def withdraw(user_id, amount):
    db.execute(
        "UPDATE accounts SET balance = balance - %s WHERE id = %s",
        amount, user_id,
    )
    db.commit()  # نجح
    kafka.produce(  # ممكن يفشل
        "balance_changed",
        {"user_id": user_id, "amount": amount},
    )

اللي بيحصل لو الـ kafka.produce فشل: الـ DB اتحدّثت، والـ event ضاع. أنت بقيت في حالة inconsistent. مفيش retry هيرجّع الـ event ده، لأن الـ caller خلص ورجع response. النتيجة: الرصيد بيقول 1500، والإيميل بيقول 2000، والـ analytics dashboard بيقول لسه 2000.

مثال للمبتدئ: مكتب البريد قبل ما يخترع حد Outbox

تخيّل إنك في مكتب بريد قديم، وعندك مهمتين كل ما يجي جواب: (1) تكتب في دفتر السجل إن الجواب اتسلّم، و(2) تحط الجواب في صندوق البريد الخارجي علشان السائق ياخده. لو كتبت في الدفتر، وقبل ما تحط الجواب في الصندوق فيه قطع كهربا، الجواب ما اتبعتش رغم إن السجل بيقول اتبعت. ده بالظبط الـ Dual Write Problem في صورة فيزيقية.

الحل الذكي: بدل ما تحط الجواب في الصندوق الخارجي مباشرة، حطه في صندوق داخلي جنب دفتر السجل. الكتابة في الدفتر وحط الجواب في الصندوق الداخلي بتحصل في نفس اللحظة (نفس الـ "transaction"). بعدين موظف تاني يجي كل شوية يفرّغ الصندوق الداخلي ويبعت الجوابات للصندوق الخارجي. لو الكهربا قطعت، الجواب لسه في الصندوق الداخلي، والموظف هيلاقيه ويبعته بعدين.

الصندوق الداخلي ده اسمه Outbox، والموظف اللي بيفرّغه اسمه Relay.

التعريف العلمي الدقيق

Outbox Pattern: بدل ما تكتب في قاعدة البيانات وتبعت event للـ message broker في عمليتين منفصلتين على شبكتين مختلفتين، تكتب الاتنين في نفس الـ ACID transaction. الـ event بيتسجّل كصف في جدول داخلي اسمه outbox داخل نفس الـ DB. بعدين عملية مستقلة (relay/dispatcher) بتقرأ الجدول ده وبتبعت الـ events لـ Kafka أو RabbitMQ، وبتعلّمها كـ delivered. الفكرة الأساسية: الـ DB transaction هي الـ source of truth الوحيد، والـ broker بقى استهلاك تالي مش جزء من الـ atomicity guarantee.

كود تنفيذي على PostgreSQL + Python

أولاً، الجدول:

CREATE TABLE outbox (
    id              BIGSERIAL PRIMARY KEY,
    aggregate_type  TEXT NOT NULL,
    aggregate_id    TEXT NOT NULL,
    event_type      TEXT NOT NULL,
    payload         JSONB NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    processed_at    TIMESTAMPTZ
);

CREATE INDEX idx_outbox_unprocessed
    ON outbox (created_at)
    WHERE processed_at IS NULL;

الـ partial index على الصفوف غير المُعالَجة بس بيخلي الـ relay يلاقيهم في O(log n) بدل ما يمشي على الجدول كله.

الكتابة في الـ application code:

def withdraw(user_id, amount):
    with db.transaction():
        db.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            amount, user_id,
        )
        db.execute(
            "INSERT INTO outbox "
            "(aggregate_type, aggregate_id, event_type, payload) "
            "VALUES (%s, %s, %s, %s::jsonb)",
            "account",
            str(user_id),
            "balance_changed",
            json.dumps({"user_id": user_id, "amount": amount}),
        )
    # خلاص. مفيش kafka.produce هنا.

دلوقتي حتى لو Kafka نزلت ساعة كاملة، الـ event محفوظ. الـ DB ضمنته جوّا الـ transaction. لو الـ UPDATE نجح والـ INSERT فشل، الـ transaction كله بيترجع. atomicity حقيقية.

الـ Relay loop:

def relay_loop():
    while True:
        rows = db.fetch(
            "SELECT id, event_type, payload FROM outbox "
            "WHERE processed_at IS NULL "
            "ORDER BY id LIMIT 100 "
            "FOR UPDATE SKIP LOCKED"
        )
        for row in rows:
            kafka.produce(row["event_type"], row["payload"])
            db.execute(
                "UPDATE outbox SET processed_at = NOW() WHERE id = %s",
                row["id"],
            )
        if not rows:
            time.sleep(0.5)

الـ FOR UPDATE SKIP LOCKED هو السحر هنا. بيخلّي عدة relay workers يشتغلوا بالتوازي بدون double-publish، لأن أي صف بياخده worker بيتقفل، والـ workers الثانيين بيتخطّوه بدل ما يستنوا. ده feature في PostgreSQL من نسخة 9.5 وفوق.

استراتيجيتين للقراءة: Polling vs CDC

عندك طريقتين تشغّل الـ relay، اختار حسب الـ latency المطلوب:

1. Polling (الكود اللي فوق): سهل، شغّال على أي DB، مفيش dependencies إضافية. الـ end-to-end latency من 100ms لـ 2 ثانية حسب فترة الـ sleep. مناسب لأكتر من 90% من الحالات.
2. Change Data Capture (CDC): أداة زي Debezium بتقرأ الـ Write-Ahead Log مباشرة (PostgreSQL logical replication slot)، وبتبعت كل INSERT في الـ outbox table لـ Kafka فوراً. الـ latency بينزل لـ 5-50ms، لكن الـ ops أصعب: محتاج Kafka Connect، replication slot مظبوط، monitoring للـ replication lag.

القاعدة: ابدأ بـ Polling. لما الـ latency يبقى bottleneck فعلي (معتاد عند معدلات أعلى من 500 event/sec أو متطلبات real-time أقل من 100ms)، انتقل لـ CDC.

أرقام مقاسة من workload إنتاج

على cluster بـ 3 خدمات بتكتب في outbox بمعدل 800 event/sec، مع PostgreSQL 16 و Kafka 3.7:

• زمن الكتابة (مع الـ INSERT في outbox): 4ms زيادة عن الكتابة العادية بدون event. مهمل.
• End-to-end latency بـ Polling كل 200ms: p50 = 230ms، p99 = 480ms.
• End-to-end latency بـ Debezium CDC: p50 = 12ms، p99 = 38ms.
• معدل الـ events المفقودة: صفر، حتى مع 4 محاكاة لانقطاع Kafka لمدة 10 دقايق كل واحدة.
• حجم الـ outbox table بعد أسبوع: 28GB قبل cleanup، 1.2GB مع cleanup يومي للصفوف اللي عمرها أكتر من 24 ساعة.

Trade-offs لازم تفهمها قبل ما تطبّق

الـ Outbox مش مجاني، ودي التكاليف الفعلية:

• Latency زيادة: الـ event ما بيوصلش فوراً. أقل ممكن مع CDC، لكن لسه فيه فرق ميلي ثواني عن الـ produce المباشر.
• تكلفة I/O على الـ DB: كل event بقى INSERT زيادة. على workload عالي جداً (أعلى من 10K event/sec)، الـ outbox بقت hot table. الحل: partitioning على created_at + cleanup job يومي.
• At-least-once delivery: الـ relay ممكن ينجح في الـ produce ويفشل في تحديث processed_at بسبب crash بين العمليتين. الـ event هيتبعت تاني عند الـ restart. علشان كده الـ consumer لازم يكون idempotent.
• ترتيب الـ events: Polling مش بيضمن FIFO عبر relay workers متعددين. لو الترتيب مهم على نفس الـ aggregate (مثلاً نفس الحساب)، استخدم relay واحد فقط، أو Debezium مع partition key ثابت = aggregate_id.

متى لا تستخدم Outbox Pattern

• لو التطبيق monolith ومفيش consumers خارجيين فعليين. الـ overhead مش مبرر.
• لو الـ event مش حرج (مثلاً analytics ضايع 0.1% مقبول). استخدم fire-and-forget مع log محلي.
• لو التطبيق بيستخدم event-sourcing بالفعل، الـ event store هو الـ outbox الطبيعي. مش محتاج جدول إضافي.
• لو محتاج strict global ordering عبر aggregates مختلفة، الـ outbox مش هيوفّره. تحتاج global sequencer أو نظام زي Kafka Streams بـ state store.

الخطوة التالية

افتح أصغر service عندك بيكتب في DB وبينشر event. ابني جدول outbox من 6 أعمدة، وغيّر سطر kafka.produce بـ INSERT داخل الـ transaction. بعد كده شغّل relay polling بسيط في process مستقل. خلال أسبوع راقب: كم event فُقد قبل التغيير؟ كم بعده؟ الفرق هو الـ ROI الفعلي اللي هيبرّر تطبيقه على باقي الخدمات.

مصادر

• Microservices.io – Pattern: Transactional outbox
• Debezium – Outbox Event Router
• PostgreSQL Documentation – SELECT FOR UPDATE SKIP LOCKED
• Chris Richardson – Microservices Patterns, Manning Publications, 2018, Chapter 3
• Apache Kafka Documentation – Producer Idempotence