> ## Content Index > Fetch the complete content index at: https://arabroot.io/llms.txt > Use this file to discover other available public pages before exploring further. # المعاملات الموزعة: Saga ونمط Outbox بدل 2PC - URL: https://arabroot.io/articles/المعاملات-الموزعة-saga-ونمط-outbox/ - Published: 2026-10-11T11:45:00.000Z - Updated: 2026-10-12T16:08:50.000Z - Description: كيف تنفذ عملية تمتد على عدة خدمات دون 2PC، بخطوات محلية وتعويضات وجدول Outbox لا يضيع فيه أي حدث. - Author: فريق عرب رووت - Tags: الأدلة التقنية, هندسة البرمجيات وتصميم الأنظمة, قواعد البيانات, DevOps وCI/CD, الاستضافة الذاتية في متجر يعمل بقاعدة بيانات واحدة، عملية «أنشئ الطلب واحجز المخزون واخصم المبلغ» معاملة Transaction واحدة، إما أن تنجح كلها أو تلغى كلها، ولا يفكر المطور في الحالة الوسطى أبداً. ولكن عندما تنقسم هذه العملية على ثلاث خدمات، لكل منها قاعدة بياناتها، تصبح الحالة الوسطى هي المشكلة الأساسية: ماذا لو حجزت خدمة المخزون القطعة ثم رفضت خدمة الدفع البطاقة؟ ومن يعيد القطعة إلى المخزون؟ وماذا لو حفظت خدمة الطلبات الطلب ثم توقفت قبل أن تخبر أحداً؟ وقد عرضنا الإجابة باختصار في [مقال الأنظمة الموزعة](https://arabroot.io/articles/%D8%A7%D9%84%D8%A7%D9%86%D8%B8%D9%85%D8%A9-%D8%A7%D9%84%D9%85%D9%88%D8%B2%D8%B9%D8%A9-%D9%84%D9%84%D9%85%D8%A8%D8%AA%D8%AF%D8%A6%D9%8A%D9%86-cap-%D9%88%D8%A7%D9%84%D8%A7%D8%AA%D8%B3%D8%A7%D9%82-%D9%88%D8%A7%D9%84%D8%A7%D8%AC%D9%85%D8%A7%D8%B9/#transactions): نتجنب الالتزام على مرحلتين 2PC، ونستخدم نمط Saga مع جدول Outbox. وفي هذا المقال نبني ذلك فعلاً، بثلاث خدمات وثلاث قواعد بيانات، ونرى الطلب الذي فشل دفعه يلغى خطوة بخطوة. [![](https://arabroot.io/content/images/2026/10/distributed-systems-cover-cover.webp)دليلالأنظمة الموزعة للمبتدئين: ماذا يحدث عندما تنقطع الشبكةلماذا تتصرف الأنظمة الموزعة بشكل غريب عند انقطاع الشبكة، وكيف تتفق السيرفرات على قرار واحد رغم ذلك.](https://arabroot.io/articles/%D8%A7%D9%84%D8%A7%D9%86%D8%B8%D9%85%D8%A9-%D8%A7%D9%84%D9%85%D9%88%D8%B2%D8%B9%D8%A9-%D9%84%D9%84%D9%85%D8%A8%D8%AA%D8%AF%D8%A6%D9%8A%D9%86-cap-%D9%88%D8%A7%D9%84%D8%A7%D8%AA%D8%B3%D8%A7%D9%82-%D9%88%D8%A7%D9%84%D8%A7%D8%AC%D9%85%D8%A7%D8%B9/)[![](https://arabroot.io/content/images/2026/10/idempotency-cover-cover.webp)دليلIdempotency: كيف تجعل العملية آمنة عند تكرارهالماذا تصل الرسالة نفسها أكثر من مرة، وكيف تكتب مستهلكاً يطبقها مرة واحدة فقط مهما تكررت.](https://arabroot.io/articles/idempotency-%D9%83%D9%8A%D9%81-%D8%AA%D8%AC%D8%B9%D9%84-%D8%A7%D9%84%D8%B9%D9%85%D9%84%D9%8A%D8%A9-%D8%A7%D9%85%D9%86%D8%A9-%D8%B9%D9%86%D8%AF-%D8%AA%D9%83%D8%B1%D8%A7%D8%B1%D9%87%D8%A7/)[![](https://arabroot.io/content/images/2026/10/monolith-microservices-cover-cover.webp)دليلMonolith أم Microservices؟ قرار صادق للفرق الصغيرة والكبيرةمتى يكفيك برنامج واحد مقسم إلى وحدات، ومتى تستحق الخدمات المصغرة ثمنها، مع حالات موثقة وتجربة تقيس الفرق.](https://arabroot.io/articles/monolith-%D8%A7%D9%85-microservices-%D9%82%D8%B1%D8%A7%D8%B1-%D8%B5%D8%A7%D8%AF%D9%82-%D9%84%D9%84%D9%81%D8%B1%D9%82-%D8%A7%D9%84%D8%B5%D8%BA%D9%8A%D8%B1%D8%A9-%D9%88%D8%A7%D9%84%D9%83%D8%A8%D9%8A%D8%B1%D8%A9/) وسوف نناقش في هذا المقال ما يلي: - لماذا تتجنب الفرق 2PC في الخدمات، وما الذي يقوله توثيق PostgreSQL عنها. - Saga بطريقتيها: الأحداث بين الخدمات Choreography والمنسق Orchestration، ومتى تختار كلاً منهما. - كيف تصمم خطوات التعويض Compensation، ولماذا ليست تراجعاً Rollback، وما الذي يراه المستخدم أثناء ذلك. - مشكلة الكتابة المزدوجة Dual Write، وجدول Outbox الذي يحلها، وال Relay الذي ينقل الأحداث بالسؤال الدوري Polling أو بقراءة سجل المعاملات CDC. - تجربة عملية على PostgreSQL وValkey: ثلاث خدمات، وطلب ينجح، وطلب يفشل دفعه فيعوض، وطلب يرفضه المخزون، وال Relay يتوقف في أسوأ لحظة. ## لماذا لا نستخدم 2PC؟ الالتزام على مرحلتين Two-Phase Commit يجعل عدة قواعد بيانات تتصرف كأنها واحدة: منسق يسأل الجميع «هل أنتم مستعدون؟» فيقفل كل مشارك صفوفه ويجيب، ثم يقول «نفذوا». وقد شرحنا في مقال الأنظمة الموزعة أن مشكلته في الفشل، فإذا توقف المنسق بين المرحلتين بقي كل مشارك قافلاً صفوفه لا يعرف القرار. ويضاف إلى ذلك ثلاثة أسباب عملية تجعل الفرق تتجنبه في الخدمات: - PostgreSQL نفسه لا يشجعك عليه، فالأمر `PREPARE TRANSACTION` كما يقول [توثيقه](https://www.postgresql.org/docs/18/sql-prepare-transaction.html?ref=arabroot.io) غير مخصص للتطبيقات، وإنما لمدير معاملات خارجي، وإذا لم تكن تكتب مدير معاملات فالأغلب أنك لا تحتاجه، والإعداد `max_prepared_transactions` قيمته الافتراضية صفر، أي أن الميزة معطلة حتى تفعلها، كما في [توثيق الإعدادات](https://www.postgresql.org/docs/18/runtime-config-resource.html?ref=arabroot.io). - أغلب ما تتحدث معه الخدمات لا يدعمه أصلاً: الطوابير، وValkey، وواجهات بوابات الدفع، ومزودي البريد. - كل الخدمات يجب أن تكون متاحة في نفس اللحظة، فإذا كانت كل خدمة متاحة 99.9% من الوقت، فالعملية التي تحتاج ثلاثاً منها معاً متاحة نحو 99.7%، وكل قفل ينتظر أبطأها. والبديل أن تقبل أن العملية الطويلة ليست معاملة واحدة، وإنما سلسلة معاملات صغيرة، لكل منها طريقة لإلغاء أثرها، وهذا هو نمط Saga. ## Saga: خطوات محلية وخطوات تعويض ال Saga سلسلة من المعاملات المحلية، كل معاملة في خدمة واحدة وقاعدة بيانات واحدة، وكل خطوة تنجح تنشر حدثاً أو رداً يبدأ الخطوة التالية، وإذا فشلت خطوة نفذت خطوات تعويض Compensating Transactions للخطوات التي نجحت قبلها بالعكس، كما يصف [نمط Saga في microservices.io](https://microservices.io/patterns/data/saga.html?ref=arabroot.io). وفي متجرنا: 1. خدمة الطلبات تنشئ الطلب بحالة PENDING. 2. خدمة المخزون تحجز القطعة، وتعويضها: فك الحجز. 3. خدمة الدفع تخصم المبلغ، وهي الخطوة الحاسمة، فإذا نجحت لا رجوع. 4. خدمة الطلبات تجعل الطلب CONFIRMED. ### بالأحداث Choreography أم بالمنسق Orchestration؟ هناك طريقتان لتنفيذ هذه السلسلة، والصورة التالية تبين الفرق: ![لوحتان. Choreography: خدمات orders وstock وpayments في مثلث، وأسهم أحداث بينها: OrderCreated من الطلبات إلى المخزون، وStockReserved من المخزون إلى الدفع، وPaymentSucceeded من الدفع إلى الطلبات. Orchestration: منسق order saga في الوسط يرسل الأوامر create وreserve وcharge إلى الخدمات الثلاث](https://arabroot.io/content/images/2026/10/saga-outbox-02-choreography.webp) في الأولى تعرف كل خدمة دورها من الأحداث، وفي الثانية يحفظ منسق واحد حالة العملية - Choreography: لا يوجد مدير، فكل خدمة تستمع إلى أحداث معينة وتنشر أحداثها، فخدمة المخزون تسمع `OrderCreated` فتحجز وتنشر `StockReserved`، وخدمة الدفع تسمع `StockReserved` فتخصم وتنشر `PaymentSucceeded`. وهي مناسبة عندما تكون الخطوات قليلة، لأن كل خدمة مستقلة ولا يوجد مكون إضافي، ولكن عندما تكبر يصعب أن تعرف المسار الكامل، فهو موزع على معالجات في عدة مستودعات، ويصبح سؤال «أين توقف الطلب رقم 3؟» بحثاً في سجلات ثلاث خدمات. - Orchestration: منسق Orchestrator يحفظ حالة كل Saga، ويرسل لكل خدمة أمراً ويستقبل ردها، ويقرر الخطوة التالية أو التعويض. والمسار كله مكتوب في مكان واحد ويسهل اختباره، والثمن أن المنسق خدمة إضافية يجب ألا تفقد حالتها. وهناك محركات مفتوحة المصدر تقوم بهذا الدور، مثل [Temporal](https://github.com/temporalio/temporal?ref=arabroot.io) برخصة MIT، الذي يحفظ حالة كل خطوة ويعيد المحاولة ويستكمل بعد التوقف. وقد يتساءل البعض: أيهما أختار؟ والإجابة العملية: ابدأ بالأحداث إذا كانت العملية ثلاث خطوات أو أربعاً بين خدمات قليلة، وانتقل إلى المنسق عندما تكثر الخطوات والفروع، أو عندما يسألك فريق الدعم كل يوم عن حالة عمليات عالقة. والتجربة العملية في هذا المقال تستخدم الأحداث لأنها أقل كوداً. ### التعويض ليس تراجعاً التراجع Rollback في قاعدة البيانات يمسح الأثر كأنه لم يحدث، ولم يره أحد. أما التعويض فهو عملية جديدة بعد أن حدث الأثر ورآه العالم: القطعة كانت محجوزة لثوان، وربما رأى عميل آخر في تلك الثواني أنها نفدت، والمبلغ إذا خصم ثم أعيد ظهر في كشف حساب العميل عمليتين. ولذلك يحتاج تصميم التعويضات إلى تفكير في العمل وليس في الكود فقط، وهذه أهم قواعده: - رتب الخطوات بحيث تأتي الخطوات القابلة للتعويض أولاً، والخطوة التي لا يمكن التراجع عنها بسهولة، مثل الخصم الفعلي أو شحن الطرد، في آخر السلسلة، فإذا فشل شيء قبلها كان التعويض سهلاً. - التعويض يجب أن ينجح في النهاية، فلا يجوز أن يفشل فك الحجز لأن الخدمة مشغولة، ولذلك يعاد حتى ينجح، ويجب أن يكون آمناً عند التكرار، كما في [مقال Idempotency](https://arabroot.io/articles/idempotency-%D9%83%D9%8A%D9%81-%D8%AA%D8%AC%D8%B9%D9%84-%D8%A7%D9%84%D8%B9%D9%85%D9%84%D9%8A%D8%A9-%D8%A7%D9%85%D9%86%D8%A9-%D8%B9%D9%86%D8%AF-%D8%AA%D9%83%D8%B1%D8%A7%D8%B1%D9%87%D8%A7/). - ال Saga لا تعطيك العزل Isolation الذي تعطيه المعاملة، فبين الخطوات يرى الآخرون حالات وسطى، ويذكر microservices.io هذا صراحة باسم غياب العزل Lack of Isolation، والحل الأبسط أن تكون الحالة الوسطى ظاهرة ومفهومة، مثل طلب بحالة PENDING لا يشحن ولا يحسب في المبيعات حتى يتأكد، وحجز في جدول منفصل بدل إنقاص المخزون النهائي. ## مشكلة الكتابة المزدوجة Dual Write كل خطوة في ال Saga تفعل شيئين: تكتب في قاعدة بياناتها، وتنشر حدثاً للخطوة التالية. والحل البسيط هو أن تكتب ثم تنشر في سطرين متتاليين، ولكن الصورة التالية تبين ما يحدث إذا توقفت الخدمة بينهما: ![لوحتان. اليسار: خدمة الطلبات تحفظ في قاعدة بياناتها بنجاح، ثم تتوقف قبل XADD إلى الوسيط، فيبقى الطلب 1 PENDING ولا يسمع به أحد. اليمين: خدمة الطلبات تدخل الطلب وصف ال outbox في معاملة واحدة، ثم يسأل ال relay الجدول وينشر إلى الوسيط ثم يعلم الصف، و0 صفوف غير منشورة في التجربة](https://arabroot.io/content/images/2026/10/saga-outbox-01-dual-write.webp) كتابتان منفصلتان تضيع بينهما الأحداث، وجدول Outbox يجعلهما كتابة واحدة إذا كتبت ثم توقفت قبل النشر بقي الطلب PENDING للأبد، ولا تعلم به خدمة المخزون، والعميل ينتظر. وإذا عكست الترتيب فنشرت ثم كتبت، ثم فشلت الكتابة، فقد أخبرت الخدمات الأخرى عن طلب غير موجود، فتحجز له قطعة. ولا يوجد ترتيب صحيح للكتابتين، لأنهما في نظامين مختلفين ولا توجد معاملة تجمعهما. والحل هو نمط [Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html?ref=arabroot.io): لا تنشر الحدث من كود العملية أبداً، وإنما تكتبه صفاً في جدول اسمه `outbox` في نفس قاعدة البيانات، داخل نفس المعاملة التي تكتب الطلب، فإما أن يحفظ الطلب وحدثه معاً أو لا يحفظ أي منهما. ثم يأتي مكون منفصل اسمه ال Relay، فيقرأ الصفوف التي لم تنشر، وينشرها إلى الوسيط، ويعلمها بأنها نشرت. فإذا توقف ال Relay قبل النشر بقي الصف وانتظره، وإذا توقف بعد النشر وقبل أن يعلم الصف فسوف ينشره مرة ثانية عند عودته، ولذلك يحتاج كل مستقبل إلى جدول منع التكرار الذي شرحناه في مقال Idempotency، وهذا ما سوف تراه في التجربة. والتطبيقات التي تستخدم Outbox على هذا الموقع كثيرة، فقد رأيته في [دليل البريد التفاعلي](https://arabroot.io/articles/%D8%A7%D9%84%D8%A8%D8%B1%D9%8A%D8%AF-%D8%A7%D9%84%D8%AA%D9%81%D8%A7%D8%B9%D9%84%D9%8A-transactional-email/) لإرسال البريد بعد التسجيل. ## كيف تخرج الأحداث من جدول Outbox؟ هناك طريقتان لبناء ال Relay، والصورة التالية تبينهما: ![لوحتان. Polling Publisher: ال relay يرسل SELECT مع SKIP LOCKED إلى جدول outbox كل 200 ملي ثانية، ثم XADD إلى الوسيط ثم UPDATE، ونقاط القوة أنه SQL فقط، ونقاط الضعف حمل السؤال الدوري وتأخير صغير وتنظيف الصفوف القديمة. Transaction Log Tailing: Debezium يقرأ WAL من PostgreSQL عبر replication slot وينشر إلى Kafka بال event router، ونقاط القوة لا يوجد سؤال دوري وتأخير قليل، ونقاط الضعف تشغيل Kafka Connect وslot عالق يبقي WAL على القرص](https://arabroot.io/content/images/2026/10/saga-outbox-03-relay.webp) السؤال الدوري يكفي أغلب التطبيقات، وقراءة سجل المعاملات للأحجام الكبيرة - السؤال الدوري [Polling Publisher](https://microservices.io/patterns/data/polling-publisher.html?ref=arabroot.io): ال Relay يسأل الجدول كل فترة قصيرة عن الصفوف التي قيمة `published_at` فيها فارغة. وهو SQL فقط ويعمل مع أي قاعدة بيانات، وتستطيع أن تشغل منه أكثر من نسخة إذا استخدمت `FOR UPDATE SKIP LOCKED`، فتأخذ كل نسخة صفوفاً لا تأخذها الأخرى. وثمنه أن الحدث يتأخر حتى السؤال التالي، وأن السؤال يعمل حتى لو لم يكن هناك شيء جديد، وهذا حمل صغير مع فهرس جزئي على الصفوف غير المنشورة كما في التجربة، وأن الصفوف المنشورة يجب أن تحذف بمهمة دورية. - قراءة سجل المعاملات [Transaction Log Tailing](https://microservices.io/patterns/data/transaction-log-tailing.html?ref=arabroot.io)، ويسمى أيضاً CDC اختصاراً لـ Change Data Capture: بدل أن تسأل الجدول، تقرأ سجل المعاملات WAL الذي يكتب فيه PostgreSQL كل تغيير، فترى كل صف جديد في جدول ال Outbox لحظة تأكيده وبترتيب حدوثه. والأداة المشهورة لذلك Debezium برخصة Apache 2.0، وفيها محول جاهز اسمه [Outbox Event Router](https://debezium.io/documentation/reference/stable/transformations/outbox-event-router.html?ref=arabroot.io) يحول صفوف الجدول إلى رسائل في Kafka. وثمنها أنك تشغل Kafka Connect وتراقبه، وأن PostgreSQL يحتفظ بال WAL حتى يقرأه Debezium عبر ما يسمى Replication Slot، ويذكر [توثيق Logical Decoding](https://www.postgresql.org/docs/18/logicaldecoding-explanation.html?ref=arabroot.io) أن ال Slot يمنع حذف ال WAL الذي يحتاجه حتى لو لم يكن أحد متصلاً به، فإذا توقف Debezium أياماً امتلأ القرص. وقد شرحنا CDC وDebezium بالتفصيل مع تجربة كاملة في مقال [نقل تغييرات قاعدة البيانات CDC مع Debezium](https://arabroot.io/articles/%D9%86%D9%82%D9%84-%D8%AA%D8%BA%D9%8A%D9%8A%D8%B1%D8%A7%D8%AA-%D9%82%D8%A7%D8%B9%D8%AF%D8%A9-%D8%A7%D9%84%D8%A8%D9%8A%D8%A7%D9%86%D8%A7%D8%AA-cdc-%D9%85%D8%B9-debezium/). والقاعدة العملية: ابدأ بالسؤال الدوري كل 100 إلى 500 ملي ثانية، فهو يكفي آلاف الأحداث في الثانية، وانتقل إلى CDC عندما يصبح التأخير أو حمل السؤال مشكلة حقيقية تقيسها، أو عندما تحتاج Kafka لأسباب أخرى. ## تجربة عملية: Saga بثلاث خدمات وجدول Outbox في هذه التجربة نشغل PostgreSQL فيه ثلاث قواعد بيانات منفصلة، `orders` و`stock` و`payments`، لكل خدمة قاعدتها ولا تقرأ خدمة جداول غيرها، وفي كل قاعدة جدول `outbox` للأحداث الخارجة وجدول `inbox` لمعرفات الأحداث التي عولجت. وValkey Streams هو الوسيط، بمجموعة مستهلكين لكل خدمة. وسكربت Python واحد يشغل الخدمات الثلاث، ولكل خدمة Relay ومستهلك في خيوط Threads منفصلة. والمخزون 5 قطع من المنتج A1 بسعر 100.00 للقطعة، وفي حساب ali مبلغ 500.00، وفي حساب sara مبلغ 20.00. المتطلبات: جهاز عليه Docker Engine وDocker Compose، ولا تحتاج نشر أي منفذ على جهازك. ### ملفات التجربة أنشئ مجلداً فيه أربعة ملفات، الأول `compose.yaml`: ```yaml name: lab-saga services: db: image: postgres:18.6 container_name: lab-saga-db environment: POSTGRES_PASSWORD: lab-only-password volumes: - ./init.sql:/docker-entrypoint-initdb.d/01-init.sql:ro healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres -d payments"] interval: 2s retries: 30 broker: image: valkey/valkey:9.1.2-alpine container_name: lab-saga-valkey command: ["valkey-server", "--appendonly", "yes"] healthcheck: test: ["CMD", "valkey-cli", "ping"] interval: 2s retries: 30 app: build: . image: lab-saga-app:1 container_name: lab-saga-app environment: PG: postgresql://postgres:lab-only-password@db:5432 VALKEY_URL: valkey://broker:6379/0 depends_on: db: condition: service_healthy broker: condition: service_healthy profiles: ["run"] ``` والثاني `init.sql`، وفيه قواعد البيانات الثلاث وجداولها: ```sql -- three services, three databases: no service reads another one's tables CREATE DATABASE orders; CREATE DATABASE stock; CREATE DATABASE payments; \c orders CREATE TABLE orders ( id bigserial PRIMARY KEY, customer text NOT NULL, sku text NOT NULL, qty int NOT NULL, amount_halala bigint NOT NULL, status text NOT NULL, -- PENDING, CONFIRMED, CANCELLED reason text ); CREATE TABLE outbox ( id bigserial PRIMARY KEY, event_id uuid NOT NULL DEFAULT gen_random_uuid(), type text NOT NULL, payload jsonb NOT NULL, created_at timestamptz NOT NULL DEFAULT now(), published_at timestamptz ); CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL; CREATE TABLE inbox (event_id uuid PRIMARY KEY, received_at timestamptz NOT NULL DEFAULT now()); \c stock CREATE TABLE stock (sku text PRIMARY KEY, qty int NOT NULL CHECK (qty >= 0)); CREATE TABLE reservations (order_id bigint PRIMARY KEY, sku text NOT NULL, qty int NOT NULL); CREATE TABLE outbox ( id bigserial PRIMARY KEY, event_id uuid NOT NULL DEFAULT gen_random_uuid(), type text NOT NULL, payload jsonb NOT NULL, created_at timestamptz NOT NULL DEFAULT now(), published_at timestamptz ); CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL; CREATE TABLE inbox (event_id uuid PRIMARY KEY, received_at timestamptz NOT NULL DEFAULT now()); \c payments CREATE TABLE accounts (customer text PRIMARY KEY, balance_halala bigint NOT NULL CHECK (balance_halala >= 0)); CREATE TABLE payments (order_id bigint PRIMARY KEY, customer text NOT NULL, amount_halala bigint NOT NULL); CREATE TABLE outbox ( id bigserial PRIMARY KEY, event_id uuid NOT NULL DEFAULT gen_random_uuid(), type text NOT NULL, payload jsonb NOT NULL, created_at timestamptz NOT NULL DEFAULT now(), published_at timestamptz ); CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL; CREATE TABLE inbox (event_id uuid PRIMARY KEY, received_at timestamptz NOT NULL DEFAULT now()); ``` في الإعداد أعلاه لاحظ التالي: - جدول `outbox` واحد في كل قاعدة، وفيه معرف الحدث `event_id` الذي يولده PostgreSQL بالدالة `gen_random_uuid()`، وهو المعرف الذي يحفظه المستقبل في جدول `inbox`. - الفهرس الجزئي `outbox_unpublished` يغطي الصفوف غير المنشورة فقط، فيبقى صغيراً مهما كبر الجدول، ويجعل سؤال ال Relay سريعاً. - القيد `CHECK (qty >= 0)` و`CHECK (balance_halala >= 0)` خط دفاع أخير، فحتى لو أخطأ الكود لن يصبح المخزون أو الرصيد سالباً. والثالث `Dockerfile`: ```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 saga.py . ENTRYPOINT ["python", "-u", "saga.py"] ``` والرابع `saga.py`: ```python # An order saga across three services (orders, stock, payments), choreographed with events. # Each service writes its change and its outgoing event in one local transaction (transactional outbox); # a relay polls the outbox and publishes to a Valkey stream; each consumer keeps an inbox to drop duplicates. # python saga.py dual-write | setup | place | run [crash-relay] | status import json, os, sys, threading, time import psycopg import valkey PG, STREAM = os.environ["PG"], "events" r = valkey.Valkey.from_url(os.environ["VALKEY_URL"], decode_responses=True) PRICE = 10000 # every unit costs 100.00 lock = threading.Lock() def db(service): return psycopg.connect(f"{PG}/{service}", autocommit=True) def log(who, text): with lock: print(f"{time.strftime('%H:%M:%S')} {who:<16} {text}", flush=True) def emit(conn, type_, payload): """Write an event to this service's outbox, inside the caller's transaction.""" conn.execute("INSERT INTO outbox (type, payload) VALUES (%s, %s)", (type_, json.dumps(payload))) # ---------------------------------------------------------------- the order service def place(customer, sku, qty): with db("orders") as c, c.transaction(): # the order and its event: one transaction oid = c.execute("INSERT INTO orders (customer, sku, qty, amount_halala, status) VALUES (%s, %s, %s, %s, 'PENDING') RETURNING id", (customer, sku, qty, qty * PRICE)).fetchone()[0] emit(c, "OrderCreated", {"order_id": oid, "customer": customer, "sku": sku, "qty": qty, "amount": qty * PRICE}) log("orders", f"order {oid} PENDING ({customer}, {qty} x {sku}) + OrderCreated in the outbox") def place_dual_write(customer, sku, qty): with db("orders") as c: oid = c.execute("INSERT INTO orders (customer, sku, qty, amount_halala, status) VALUES (%s, %s, %s, %s, 'PENDING') RETURNING id", (customer, sku, qty, qty * PRICE)).fetchone()[0] log("orders", f"order {oid} committed to the orders database") log("orders", "process killed (deploy, OOM) before XADD OrderCreated") os._exit(1) # the second write never happens r.xadd(STREAM, {"type": "OrderCreated"}) def on_orders(c, ev, p): if ev == "PaymentSucceeded": c.execute("UPDATE orders SET status = 'CONFIRMED' WHERE id = %s AND status = 'PENDING'", (p["order_id"],)) return f"order {p['order_id']} CONFIRMED" if ev == "StockRejected": c.execute("UPDATE orders SET status = 'CANCELLED', reason = 'out of stock' WHERE id = %s AND status = 'PENDING'", (p["order_id"],)) return f"order {p['order_id']} CANCELLED (out of stock)" if ev == "StockReleased": # the end of the compensation chain c.execute("UPDATE orders SET status = 'CANCELLED', reason = 'payment failed' WHERE id = %s AND status = 'PENDING'", (p["order_id"],)) return f"order {p['order_id']} CANCELLED (payment failed, stock released)" # ---------------------------------------------------------------- the stock service def on_stock(c, ev, p): if ev == "OrderCreated": ok = c.execute("UPDATE stock SET qty = qty - %s WHERE sku = %s AND qty >= %s", (p["qty"], p["sku"], p["qty"])).rowcount if ok: c.execute("INSERT INTO reservations VALUES (%s, %s, %s)", (p["order_id"], p["sku"], p["qty"])) emit(c, "StockReserved", p) return f"reserved {p['qty']} x {p['sku']} for order {p['order_id']} -> StockReserved" emit(c, "StockRejected", p) return f"not enough {p['sku']} for order {p['order_id']} -> StockRejected" if ev == "PaymentFailed": # compensation: undo our own earlier step row = c.execute("DELETE FROM reservations WHERE order_id = %s RETURNING sku, qty", (p["order_id"],)).fetchone() if row: c.execute("UPDATE stock SET qty = qty + %s WHERE sku = %s", (row[1], row[0])) emit(c, "StockReleased", p) return f"COMPENSATION: released {row[1] if row else 0} x {p['sku']} of order {p['order_id']} -> StockReleased" # ---------------------------------------------------------------- the payment service def on_payments(c, ev, p): if ev == "StockReserved": ok = c.execute("UPDATE accounts SET balance_halala = balance_halala - %s WHERE customer = %s AND balance_halala >= %s", (p["amount"], p["customer"], p["amount"])).rowcount if ok: c.execute("INSERT INTO payments VALUES (%s, %s, %s)", (p["order_id"], p["customer"], p["amount"])) emit(c, "PaymentSucceeded", p) return f"charged {p['customer']} {p['amount'] / 100:.2f} for order {p['order_id']} -> PaymentSucceeded" emit(c, "PaymentFailed", p) return f"{p['customer']} cannot pay {p['amount'] / 100:.2f} for order {p['order_id']} -> PaymentFailed" HANDLERS = {"orders": on_orders, "stock": on_stock, "payments": on_payments} LISTENS = {"orders": {"PaymentSucceeded", "StockRejected", "StockReleased"}, "stock": {"OrderCreated", "PaymentFailed"}, "payments": {"StockReserved"}} # ---------------------------------------------------------------- relay: outbox -> stream (polling publisher) def relay(service, stop, crash_once): conn = db(service) while not stop.is_set(): try: with conn.transaction(): rows = conn.execute("""SELECT id, event_id, type, payload FROM outbox WHERE published_at IS NULL ORDER BY id LIMIT 50 FOR UPDATE SKIP LOCKED""").fetchall() for oid, eid, type_, payload in rows: r.xadd(STREAM, {"event_id": str(eid), "type": type_, "source": service, "payload": json.dumps(payload)}) if rows and crash_once.is_set(): crash_once.clear() raise RuntimeError("relay crashed after XADD, before marking the rows") if rows: conn.execute("UPDATE outbox SET published_at = now() WHERE id = ANY(%s)", ([x[0] for x in rows],)) log(f"relay/{service}", f"published {', '.join(x[2] for x in rows)}") except RuntimeError as e: log(f"relay/{service}", f"{e}; restarting") time.sleep(0.2) # the polling interval conn.close() # ---------------------------------------------------------------- consumer: stream -> handler, with an inbox def consumer(service, stop): conn, handler = db(service), HANDLERS[service] while not stop.is_set(): for _, msgs in r.xreadgroup(service, service, {STREAM: ">"}, count=20, block=300) or []: for sid, m in msgs: if m["type"] not in LISTENS[service]: # not for this service r.xack(STREAM, service, sid); continue p = json.loads(m["payload"]) with conn.transaction(): # inbox row + business change + outgoing event: one transaction first = conn.execute("INSERT INTO inbox (event_id) VALUES (%s) ON CONFLICT DO NOTHING", (m["event_id"],)).rowcount result = handler(conn, m["type"], p) if first else None if not first: log(service, f"duplicate {m['type']} for order {p['order_id']} ({m['event_id'][:8]}), skipped") elif result: log(service, result) r.xack(STREAM, service, sid) conn.close() def run(seconds, crash_relay): stop, crash = threading.Event(), threading.Event() if crash_relay: crash.set() threads = [threading.Thread(target=relay, args=(s, stop, crash if s == "orders" else threading.Event())) for s in HANDLERS] threads += [threading.Thread(target=consumer, args=(s, stop)) for s in HANDLERS] [t.start() for t in threads] time.sleep(seconds); stop.set(); [t.join() for t in threads] def setup(): r.delete(STREAM) for s in HANDLERS: r.xgroup_create(STREAM, s, id="0", mkstream=True) with db("stock") as c: c.execute("INSERT INTO stock VALUES ('A1', 5)") with db("payments") as c: c.execute("INSERT INTO accounts VALUES ('ali', 50000), ('sara', 2000)") log("setup", "stock A1 = 5, ali has 500.00, sara has 20.00") def status(): with db("orders") as c: print("orders:") for row in c.execute("SELECT id, customer, qty, sku, amount_halala / 100.0, status, coalesce(reason, '') FROM orders ORDER BY id"): print(" #%s %-5s %2s x %s %8.2f %-9s %s" % row) print(" unpublished outbox rows:", c.execute("SELECT count(*) FROM outbox WHERE published_at IS NULL").fetchone()[0]) with db("stock") as c: print("stock:", c.execute("SELECT qty FROM stock WHERE sku = 'A1'").fetchone()[0], "x A1 reservations:", c.execute("SELECT coalesce(json_agg(order_id), '[]') FROM reservations").fetchone()[0]) with db("payments") as c: print("payments:", ", ".join(f"{a} {b / 100:.2f}" for a, b in c.execute("SELECT customer, balance_halala FROM accounts ORDER BY 1")), " charged orders:", c.execute("SELECT coalesce(json_agg(order_id), '[]') FROM payments").fetchone()[0]) cmd = sys.argv[1] if cmd == "setup": setup() elif cmd == "dual-write": place_dual_write(sys.argv[2], sys.argv[3], int(sys.argv[4])) elif cmd == "place": place(sys.argv[2], sys.argv[3], int(sys.argv[4])) elif cmd == "run": run(float(sys.argv[2]), len(sys.argv) > 3 and sys.argv[3] == "crash-relay") elif cmd == "status": status() ``` في الكود أعلاه لاحظ التالي: - الدالة `emit` لا ترسل شيئاً إلى الوسيط، وإنما تدخل صفاً في ال `outbox` بنفس الاتصال، فيصبح الحدث جزءاً من معاملة المستدعي. - في `consumer` يدخل معرف الحدث في `inbox`، ويعدل الجدول، ويكتب الحدث التالي في ال `outbox`، والثلاثة في معاملة واحدة، ثم يرسل `XACK` بعد COMMIT. - خطوة التعويض في `on_stock` عند وصول `PaymentFailed` تحذف الحجز وتعيد الكمية، ثم تنشر `StockReleased`، فتلغي خدمة الطلبات الطلب عندما تسمعه، أي أن الطلب لا يلغى قبل أن يعود المخزون فعلاً. - كل تحديث للحالة مشروط بالحالة السابقة، مثل `WHERE id = %s AND status = 'PENDING'`، فالحدث المكرر أو المتأخر لا يغير طلباً انتهى. - الخيار `crash-relay` يجعل Relay خدمة الطلبات ينشر الأحداث ثم يفشل قبل أن يعلم الصفوف، فتتراجع معاملته وتبقى الصفوف غير منشورة، وهذا أسوأ توقيت ممكن لل Relay. الآن نشغل PostgreSQL وValkey ونبني صورة الخدمات ونجهز البيانات: ```bash docker compose up -d --wait docker compose --profile run build app docker compose run --rm -q app setup ``` والمخرج سوف يكون كما يلي: ```text 05:19:31 setup stock A1 = 5, ali has 500.00, sara has 20.00 ``` ### الكتابة المزدوجة تضيع طلباً أولاً ننشئ طلباً بالطريقة البسيطة، أي نحفظه ثم ننشر الحدث، ونوقف العملية بينهما: ```bash docker compose run --rm -q app dual-write ali A1 1 docker compose run --rm -q app status ``` والمخرج سوف يكون كما يلي: ```text 05:19:35 orders order 1 committed to the orders database 05:19:35 orders process killed (deploy, OOM) before XADD OrderCreated orders: #1 ali 1 x A1 100.00 PENDING unpublished outbox rows: 0 stock: 5 x A1 reservations: [] payments: ali 500.00, sara 20.00 charged orders: [] ``` الطلب 1 محفوظ بحالة PENDING، ولا يوجد له حدث في أي مكان، لا في الوسيط ولا في ال `outbox`، فلن تسمع به أي خدمة، وسوف يبقى معلقاً حتى يكتشفه إنسان، وسيبقى كذلك حتى نهاية التجربة. ### ثلاثة طلبات مع Outbox الآن ننشئ ثلاثة طلبات بالطريقة الصحيحة: طلب ali لقطعتين، وطلب sara لقطعة واحدة ورصيدها لا يكفي، وطلب ali لعشر قطع والمخزون خمس: ```bash docker compose run --rm -q app place ali A1 2 docker compose run --rm -q app place sara A1 1 docker compose run --rm -q app place ali A1 10 docker compose run --rm -q app status ``` والمخرج سوف يكون كما يلي: ```text 05:19:49 orders order 2 PENDING (ali, 2 x A1) + OrderCreated in the outbox 05:19:55 orders order 3 PENDING (sara, 1 x A1) + OrderCreated in the outbox 05:20:02 orders order 4 PENDING (ali, 10 x A1) + OrderCreated in the outbox orders: #1 ali 1 x A1 100.00 PENDING #2 ali 2 x A1 200.00 PENDING #3 sara 1 x A1 100.00 PENDING #4 ali 10 x A1 1000.00 PENDING unpublished outbox rows: 3 stock: 5 x A1 reservations: [] payments: ali 500.00, sara 20.00 charged orders: [] ``` لم تعمل أي خدمة أخرى بعد، ولكن الأحداث الثلاثة محفوظة في ال `outbox` تنتظر، فلو توقف كل شيء الآن لن يضيع شيء. ### تشغيل الخدمات، وال Relay يتوقف في أسوأ لحظة الآن نشغل الخدمات الثلاث لمدة أربع ثوان، مع الخيار الذي يجعل Relay خدمة الطلبات يفشل بعد النشر وقبل تعليم الصفوف مرة واحدة: ```bash docker compose run --rm -q app run 4 crash-relay docker compose run --rm -q app status ``` والمخرج سوف يكون كما يلي: ```text 05:20:13 relay/orders relay crashed after XADD, before marking the rows; restarting 05:20:13 stock reserved 2 x A1 for order 2 -> StockReserved 05:20:13 stock reserved 1 x A1 for order 3 -> StockReserved 05:20:13 stock not enough A1 for order 4 -> StockRejected 05:20:13 relay/orders published OrderCreated, OrderCreated, OrderCreated 05:20:13 relay/stock published StockReserved, StockReserved, StockRejected 05:20:13 stock duplicate OrderCreated for order 2 (1d6ec202), skipped 05:20:13 stock duplicate OrderCreated for order 3 (b6836de9), skipped 05:20:13 stock duplicate OrderCreated for order 4 (0c92aa13), skipped 05:20:13 payments charged ali 200.00 for order 2 -> PaymentSucceeded 05:20:13 orders order 4 CANCELLED (out of stock) 05:20:13 payments sara cannot pay 100.00 for order 3 -> PaymentFailed 05:20:13 relay/payments published PaymentSucceeded, PaymentFailed 05:20:13 stock COMPENSATION: released 1 x A1 of order 3 -> StockReleased 05:20:13 orders order 2 CONFIRMED 05:20:13 relay/stock published StockReleased 05:20:13 orders order 3 CANCELLED (payment failed, stock released) orders: #1 ali 1 x A1 100.00 PENDING #2 ali 2 x A1 200.00 CONFIRMED #3 sara 1 x A1 100.00 CANCELLED payment failed #4 ali 10 x A1 1000.00 CANCELLED out of stock unpublished outbox rows: 0 stock: 3 x A1 reservations: [2] payments: ali 300.00, sara 20.00 charged orders: [2] ``` في المخرج أعلاه لاحظ التالي: - ال Relay نشر الأحداث الثلاثة ثم فشل، فتراجعت معاملته وبقيت الصفوف غير منشورة، وفي المحاولة التالية نشرها مرة ثانية، فوصل كل حدث `OrderCreated` مرتين إلى خدمة المخزون. - خدمة المخزون عالجت النسخ الأولى، ثم وجدت معرفات النسخ الثانية في جدول `inbox` فتجاهلتها، فلم تحجز أي قطعة مرتين، ولولا هذا الجدول لحجزت للطلب 2 أربع قطع. - الطلب 4 رفضه المخزون، فلم يصل إلى خدمة الدفع أصلاً، وألغي مباشرة. - الطلب 3 حجز قطعة، ثم رفضت خدمة الدفع لأن رصيد sara 20.00، فنفذت خدمة المخزون التعويض وأعادت القطعة، ثم ألغي الطلب، والصورة التالية تبين هذا المسار. - الحالة النهائية متسقة في الخدمات الثلاث: المخزون 3 لأن الطلب 2 حجز قطعتين فقط، والحجز الوحيد الباقي للطلب 2، والمبلغ الوحيد المخصوم 200.00 من ali. - الطلب 1 من الكتابة المزدوجة ما زال PENDING، فلا شيء في النظام يعرف أنه موجود. ![ثلاثة مسارات أفقية للخدمات orders وstock وpayments، وخمس خطوات مرقمة للطلب 3: خدمة الطلبات تحفظ الطلب PENDING مع ال outbox، والحدث OrderCreated إلى المخزون الذي يحجز قطعة، والحدث StockReserved إلى الدفع الذي يرفض لأن الرصيد لا يكفي، والحدث PaymentFailed يعود إلى المخزون الذي يفك الحجز، والحدث StockReleased إلى الطلبات التي تلغي الطلب. وتحتها عبارة التعويض ليس تراجعاً](https://arabroot.io/content/images/2026/10/saga-outbox-04-flow.webp) مسار الطلب 3 في التجربة: كل خطوة معاملة محلية، والحجز يلغى بخطوة تعويض ولكي ترى جدول ال Outbox في خدمة الطلبات بعد التجربة: ```bash docker exec lab-saga-db psql -U postgres -d orders -c "SELECT id, left(event_id::text, 8) AS event_id, type, payload->>'order_id' AS order_id, published_at IS NOT NULL AS published FROM outbox ORDER BY id" ``` والمخرج سوف يكون كما يلي: ```text id | event_id | type | order_id | published ----+----------+--------------+----------+----------- 1 | 1d6ec202 | OrderCreated | 2 | t 2 | b6836de9 | OrderCreated | 3 | t 3 | 0c92aa13 | OrderCreated | 4 | t (3 rows) ``` ثلاثة صفوف فقط رغم أن كل حدث نشر مرتين، ومعرفاتها هي نفسها التي ظهرت في سطور `duplicate`، أما في الوسيط فعدد الرسائل في ال Stream كان 12 رسالة بالأمر `XLEN events`: 3 أحداث مكررة، و9 أحداث للمسار نفسه. ### حذف التجربة ```bash docker compose --profile run down -v docker image rm lab-saga-app:1 ``` ## أخطاء شائعة - حفظ التغيير ثم نشر الحدث من نفس الكود، والاعتماد على أن العملية لن تتوقف بين السطرين. - Relay دون مستقبل آمن عند التكرار، فكل توقف لل Relay بعد النشر يكرر أثراً. - تعويض قد يفشل نهائياً، مثل فك حجز يرفض لأن الحجز انتهت مدته، دون أن يعامل ذلك كنجاح. - الخطوة التي لا رجوع عنها في بداية السلسلة بدل آخرها. - عرض الحالة الوسطى للمستخدم كأنها نهائية، مثل رسالة «تم الطلب» قبل أن يتأكد الدفع. - جدول Outbox لا يحذف منه شيء، أو Replication Slot متروك بعد إيقاف Debezium. - 2PC بين الخدمات لأن «المعاملة أبسط»، ثم صفوف مقفلة بعد أول توقف للمنسق. - عدم وجود مراقبة للطلبات العالقة في حالة وسطى أكثر من مدة معقولة، وهي الطريقة الوحيدة لاكتشاف الطلب 1 في التجربة. ## الخلاصة وصلنا لنهاية الموضوع، وأهم ما فيه: - 2PC يقفل الجميع خلف منسق قد يتوقف، وPostgreSQL يعطله افتراضياً ويقول إنه ليس للتطبيقات، فالبديل سلسلة معاملات محلية هي ال Saga. - الأحداث Choreography أبسط للخطوات القليلة، والمنسق Orchestration أوضح عندما تكثر الخطوات. - التعويض عملية جديدة يراها العالم وليس تراجعاً، فضع الخطوة الحاسمة آخراً، واجعل الحالات الوسطى ظاهرة. - الكتابة ثم النشر تضيع أحداثاً، وفي التجربة بقي الطلب 1 معلقاً للأبد، وجدول Outbox في نفس المعاملة يحل ذلك. - ال Relay قد ينشر الحدث مرتين، وجدول inbox عند المستقبل جعل التكرار دون أثر. - السؤال الدوري يكفي أغلب التطبيقات، وCDC بأداة مثل Debezium للأحجام الكبيرة مع مراقبة ال Replication Slot. وفي مقالات هذا القسم ما يكمل هذا الموضوع، فجعل كل خطوة آمنة عند التكرار شرط لكل ما سبق، والتخزين المؤقت وحد المعدل يحميان الخدمات من الحمل: [![](https://arabroot.io/content/images/2026/10/idempotency-cover-cover.webp)دليلIdempotency: كيف تجعل العملية آمنة عند تكرارهالماذا تصل الرسالة نفسها أكثر من مرة، وكيف تكتب مستهلكاً يطبقها مرة واحدة فقط مهما تكررت.](https://arabroot.io/articles/idempotency-%D9%83%D9%8A%D9%81-%D8%AA%D8%AC%D8%B9%D9%84-%D8%A7%D9%84%D8%B9%D9%85%D9%84%D9%8A%D8%A9-%D8%A7%D9%85%D9%86%D8%A9-%D8%B9%D9%86%D8%AF-%D8%AA%D9%83%D8%B1%D8%A7%D8%B1%D9%87%D8%A7/)[![](https://arabroot.io/content/images/2026/10/caching-strategies-cover-cover.webp)دليلالتخزين المؤقت Caching بالتفصيل: الاستراتيجيات والإبطالكيف تختار نمط التخزين المؤقت المناسب، ومتى تبطل القيمة، وكيف تمنع انهيار قاعدة البيانات عندما ينتهي مفتاح مشهور.](https://arabroot.io/articles/%D8%A7%D9%84%D8%AA%D8%AE%D8%B2%D9%8A%D9%86-%D8%A7%D9%84%D9%85%D8%A4%D9%82%D8%AA-caching-%D8%A7%D9%84%D8%A7%D8%B3%D8%AA%D8%B1%D8%A7%D8%AA%D9%8A%D8%AC%D9%8A%D8%A7%D8%AA-%D9%88%D8%A7%D9%84%D8%A7%D8%A8%D8%B7%D8%A7%D9%84/)[![](https://arabroot.io/content/images/2026/10/rate-limiting-cover-cover.webp)دليلتحديد معدل الطلبات Rate Limiting وخوارزمياتهكيف تعمل خوارزميات تحديد معدل الطلبات، وأيها يناسب واجهتك، وكيف تطبقها على عدة سيرفرات وترد بالرمز 429 الصحيح.](https://arabroot.io/articles/%D8%AA%D8%AD%D8%AF%D9%8A%D8%AF-%D9%85%D8%B9%D8%AF%D9%84-%D8%A7%D9%84%D8%B7%D9%84%D8%A8%D8%A7%D8%AA-rate-limiting-%D9%88%D8%AE%D9%88%D8%A7%D8%B1%D8%B2%D9%85%D9%8A%D8%A7%D8%AA%D9%87/)[![](https://arabroot.io/content/images/2026/10/queue-vs-streaming-cover-cover.webp)دليلالطابور Queue أم بث الأحداث Event Streaming؟ما يحدث للرسالة بعد قراءتها هو الفرق بين طابور مثل RabbitMQ وسجل أحداث مثل Kafka، ومنه تعرف أيهما يناسب مشروعك.](https://arabroot.io/articles/%D8%A7%D9%84%D8%B7%D8%A7%D8%A8%D9%88%D8%B1-%D8%A7%D9%85-%D8%A8%D8%AB-%D8%A7%D9%84%D8%A7%D8%AD%D8%AF%D8%A7%D8%AB-message-queue-%D9%88-event-streaming/) ## سجل التحديثات - أكتوبر 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. - أكتوبر 2026: رابط إلى مقال Debezium الذي يشرح قراءة سجل المعاملات CDC بالتفصيل.