في متجر يعمل بقاعدة بيانات واحدة، عملية «أنشئ الطلب واحجز المخزون واخصم المبلغ» معاملة Transaction واحدة، إما أن تنجح كلها أو تلغى كلها، ولا يفكر المطور في الحالة الوسطى أبداً. ولكن عندما تنقسم هذه العملية على ثلاث خدمات، لكل منها قاعدة بياناتها، تصبح الحالة الوسطى هي المشكلة الأساسية: ماذا لو حجزت خدمة المخزون القطعة ثم رفضت خدمة الدفع البطاقة؟ ومن يعيد القطعة إلى المخزون؟ وماذا لو حفظت خدمة الطلبات الطلب ثم توقفت قبل أن تخبر أحداً؟ وقد عرضنا الإجابة باختصار في مقال الأنظمة الموزعة: نتجنب الالتزام على مرحلتين 2PC، ونستخدم نمط Saga مع جدول Outbox. وفي هذا المقال نبني ذلك فعلاً، بثلاث خدمات وثلاث قواعد بيانات، ونرى الطلب الذي فشل دفعه يلغى خطوة بخطوة.
دليلالأنظمة الموزعة للمبتدئين: ماذا يحدث عندما تنقطع الشبكةلماذا تتصرف الأنظمة الموزعة بشكل غريب عند انقطاع الشبكة، وكيف تتفق السيرفرات على قرار واحد رغم ذلك.
دليلIdempotency: كيف تجعل العملية آمنة عند تكرارهالماذا تصل الرسالة نفسها أكثر من مرة، وكيف تكتب مستهلكاً يطبقها مرة واحدة فقط مهما تكررت.
دليلMonolith أم Microservices؟ قرار صادق للفرق الصغيرة والكبيرةمتى يكفيك برنامج واحد مقسم إلى وحدات، ومتى تستحق الخدمات المصغرة ثمنها، مع حالات موثقة وتجربة تقيس الفرق.وسوف نناقش في هذا المقال ما يلي:
- لماذا تتجنب الفرق 2PC في الخدمات، وما الذي يقوله توثيق PostgreSQL عنها.
- Saga بطريقتيها: الأحداث بين الخدمات Choreography والمنسق Orchestration، ومتى تختار كلاً منهما.
- كيف تصمم خطوات التعويض Compensation، ولماذا ليست تراجعاً Rollback، وما الذي يراه المستخدم أثناء ذلك.
- مشكلة الكتابة المزدوجة Dual Write، وجدول Outbox الذي يحلها، وال Relay الذي ينقل الأحداث بالسؤال الدوري Polling أو بقراءة سجل المعاملات CDC.
- تجربة عملية على PostgreSQL وValkey: ثلاث خدمات، وطلب ينجح، وطلب يفشل دفعه فيعوض، وطلب يرفضه المخزون، وال Relay يتوقف في أسوأ لحظة.
لماذا لا نستخدم 2PC؟
الالتزام على مرحلتين Two-Phase Commit يجعل عدة قواعد بيانات تتصرف كأنها واحدة: منسق يسأل الجميع «هل أنتم مستعدون؟» فيقفل كل مشارك صفوفه ويجيب، ثم يقول «نفذوا». وقد شرحنا في مقال الأنظمة الموزعة أن مشكلته في الفشل، فإذا توقف المنسق بين المرحلتين بقي كل مشارك قافلاً صفوفه لا يعرف القرار. ويضاف إلى ذلك ثلاثة أسباب عملية تجعل الفرق تتجنبه في الخدمات:
- PostgreSQL نفسه لا يشجعك عليه، فالأمر
PREPARE TRANSACTIONكما يقول توثيقه غير مخصص للتطبيقات، وإنما لمدير معاملات خارجي، وإذا لم تكن تكتب مدير معاملات فالأغلب أنك لا تحتاجه، والإعدادmax_prepared_transactionsقيمته الافتراضية صفر، أي أن الميزة معطلة حتى تفعلها، كما في توثيق الإعدادات. - أغلب ما تتحدث معه الخدمات لا يدعمه أصلاً: الطوابير، وValkey، وواجهات بوابات الدفع، ومزودي البريد.
- كل الخدمات يجب أن تكون متاحة في نفس اللحظة، فإذا كانت كل خدمة متاحة 99.9% من الوقت، فالعملية التي تحتاج ثلاثاً منها معاً متاحة نحو 99.7%، وكل قفل ينتظر أبطأها.
والبديل أن تقبل أن العملية الطويلة ليست معاملة واحدة، وإنما سلسلة معاملات صغيرة، لكل منها طريقة لإلغاء أثرها، وهذا هو نمط Saga.
Saga: خطوات محلية وخطوات تعويض
ال Saga سلسلة من المعاملات المحلية، كل معاملة في خدمة واحدة وقاعدة بيانات واحدة، وكل خطوة تنجح تنشر حدثاً أو رداً يبدأ الخطوة التالية، وإذا فشلت خطوة نفذت خطوات تعويض Compensating Transactions للخطوات التي نجحت قبلها بالعكس، كما يصف نمط Saga في microservices.io. وفي متجرنا:
- خدمة الطلبات تنشئ الطلب بحالة PENDING.
- خدمة المخزون تحجز القطعة، وتعويضها: فك الحجز.
- خدمة الدفع تخصم المبلغ، وهي الخطوة الحاسمة، فإذا نجحت لا رجوع.
- خدمة الطلبات تجعل الطلب CONFIRMED.
بالأحداث Choreography أم بالمنسق Orchestration؟
هناك طريقتان لتنفيذ هذه السلسلة، والصورة التالية تبين الفرق:
- Choreography: لا يوجد مدير، فكل خدمة تستمع إلى أحداث معينة وتنشر أحداثها، فخدمة المخزون تسمع
OrderCreatedفتحجز وتنشرStockReserved، وخدمة الدفع تسمعStockReservedفتخصم وتنشرPaymentSucceeded. وهي مناسبة عندما تكون الخطوات قليلة، لأن كل خدمة مستقلة ولا يوجد مكون إضافي، ولكن عندما تكبر يصعب أن تعرف المسار الكامل، فهو موزع على معالجات في عدة مستودعات، ويصبح سؤال «أين توقف الطلب رقم 3؟» بحثاً في سجلات ثلاث خدمات. - Orchestration: منسق Orchestrator يحفظ حالة كل Saga، ويرسل لكل خدمة أمراً ويستقبل ردها، ويقرر الخطوة التالية أو التعويض. والمسار كله مكتوب في مكان واحد ويسهل اختباره، والثمن أن المنسق خدمة إضافية يجب ألا تفقد حالتها. وهناك محركات مفتوحة المصدر تقوم بهذا الدور، مثل Temporal برخصة MIT، الذي يحفظ حالة كل خطوة ويعيد المحاولة ويستكمل بعد التوقف.
وقد يتساءل البعض: أيهما أختار؟ والإجابة العملية: ابدأ بالأحداث إذا كانت العملية ثلاث خطوات أو أربعاً بين خدمات قليلة، وانتقل إلى المنسق عندما تكثر الخطوات والفروع، أو عندما يسألك فريق الدعم كل يوم عن حالة عمليات عالقة. والتجربة العملية في هذا المقال تستخدم الأحداث لأنها أقل كوداً.
التعويض ليس تراجعاً
التراجع Rollback في قاعدة البيانات يمسح الأثر كأنه لم يحدث، ولم يره أحد. أما التعويض فهو عملية جديدة بعد أن حدث الأثر ورآه العالم: القطعة كانت محجوزة لثوان، وربما رأى عميل آخر في تلك الثواني أنها نفدت، والمبلغ إذا خصم ثم أعيد ظهر في كشف حساب العميل عمليتين. ولذلك يحتاج تصميم التعويضات إلى تفكير في العمل وليس في الكود فقط، وهذه أهم قواعده:
- رتب الخطوات بحيث تأتي الخطوات القابلة للتعويض أولاً، والخطوة التي لا يمكن التراجع عنها بسهولة، مثل الخصم الفعلي أو شحن الطرد، في آخر السلسلة، فإذا فشل شيء قبلها كان التعويض سهلاً.
- التعويض يجب أن ينجح في النهاية، فلا يجوز أن يفشل فك الحجز لأن الخدمة مشغولة، ولذلك يعاد حتى ينجح، ويجب أن يكون آمناً عند التكرار، كما في مقال Idempotency.
- ال Saga لا تعطيك العزل Isolation الذي تعطيه المعاملة، فبين الخطوات يرى الآخرون حالات وسطى، ويذكر microservices.io هذا صراحة باسم غياب العزل Lack of Isolation، والحل الأبسط أن تكون الحالة الوسطى ظاهرة ومفهومة، مثل طلب بحالة PENDING لا يشحن ولا يحسب في المبيعات حتى يتأكد، وحجز في جدول منفصل بدل إنقاص المخزون النهائي.
مشكلة الكتابة المزدوجة Dual Write
كل خطوة في ال Saga تفعل شيئين: تكتب في قاعدة بياناتها، وتنشر حدثاً للخطوة التالية. والحل البسيط هو أن تكتب ثم تنشر في سطرين متتاليين، ولكن الصورة التالية تبين ما يحدث إذا توقفت الخدمة بينهما:
إذا كتبت ثم توقفت قبل النشر بقي الطلب PENDING للأبد، ولا تعلم به خدمة المخزون، والعميل ينتظر. وإذا عكست الترتيب فنشرت ثم كتبت، ثم فشلت الكتابة، فقد أخبرت الخدمات الأخرى عن طلب غير موجود، فتحجز له قطعة. ولا يوجد ترتيب صحيح للكتابتين، لأنهما في نظامين مختلفين ولا توجد معاملة تجمعهما.
والحل هو نمط Transactional Outbox: لا تنشر الحدث من كود العملية أبداً، وإنما تكتبه صفاً في جدول اسمه outbox في نفس قاعدة البيانات، داخل نفس المعاملة التي تكتب الطلب، فإما أن يحفظ الطلب وحدثه معاً أو لا يحفظ أي منهما. ثم يأتي مكون منفصل اسمه ال Relay، فيقرأ الصفوف التي لم تنشر، وينشرها إلى الوسيط، ويعلمها بأنها نشرت. فإذا توقف ال Relay قبل النشر بقي الصف وانتظره، وإذا توقف بعد النشر وقبل أن يعلم الصف فسوف ينشره مرة ثانية عند عودته، ولذلك يحتاج كل مستقبل إلى جدول منع التكرار الذي شرحناه في مقال Idempotency، وهذا ما سوف تراه في التجربة. والتطبيقات التي تستخدم Outbox على هذا الموقع كثيرة، فقد رأيته في دليل البريد التفاعلي لإرسال البريد بعد التسجيل.
كيف تخرج الأحداث من جدول Outbox؟
هناك طريقتان لبناء ال Relay، والصورة التالية تبينهما:
- السؤال الدوري Polling Publisher: ال Relay يسأل الجدول كل فترة قصيرة عن الصفوف التي قيمة
published_atفيها فارغة. وهو SQL فقط ويعمل مع أي قاعدة بيانات، وتستطيع أن تشغل منه أكثر من نسخة إذا استخدمتFOR UPDATE SKIP LOCKED، فتأخذ كل نسخة صفوفاً لا تأخذها الأخرى. وثمنه أن الحدث يتأخر حتى السؤال التالي، وأن السؤال يعمل حتى لو لم يكن هناك شيء جديد، وهذا حمل صغير مع فهرس جزئي على الصفوف غير المنشورة كما في التجربة، وأن الصفوف المنشورة يجب أن تحذف بمهمة دورية. - قراءة سجل المعاملات Transaction Log Tailing، ويسمى أيضاً CDC اختصاراً لـ Change Data Capture: بدل أن تسأل الجدول، تقرأ سجل المعاملات WAL الذي يكتب فيه PostgreSQL كل تغيير، فترى كل صف جديد في جدول ال Outbox لحظة تأكيده وبترتيب حدوثه. والأداة المشهورة لذلك Debezium برخصة Apache 2.0، وفيها محول جاهز اسمه Outbox Event Router يحول صفوف الجدول إلى رسائل في Kafka. وثمنها أنك تشغل Kafka Connect وتراقبه، وأن PostgreSQL يحتفظ بال WAL حتى يقرأه Debezium عبر ما يسمى Replication Slot، ويذكر توثيق Logical Decoding أن ال Slot يمنع حذف ال WAL الذي يحتاجه حتى لو لم يكن أحد متصلاً به، فإذا توقف Debezium أياماً امتلأ القرص. وقد شرحنا CDC وDebezium بالتفصيل مع تجربة كاملة في مقال نقل تغييرات قاعدة البيانات CDC مع 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:
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، وفيه قواعد البيانات الثلاث وجداولها:
-- 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:
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:
# 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 <customer> <sku> <qty> | run <seconds> [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 ونبني صورة الخدمات ونجهز البيانات:
docker compose up -d --wait
docker compose --profile run build app
docker compose run --rm -q app setupوالمخرج سوف يكون كما يلي:
05:19:31 setup stock A1 = 5, ali has 500.00, sara has 20.00الكتابة المزدوجة تضيع طلباً
أولاً ننشئ طلباً بالطريقة البسيطة، أي نحفظه ثم ننشر الحدث، ونوقف العملية بينهما:
docker compose run --rm -q app dual-write ali A1 1
docker compose run --rm -q app statusوالمخرج سوف يكون كما يلي:
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 لعشر قطع والمخزون خمس:
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والمخرج سوف يكون كما يلي:
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 خدمة الطلبات يفشل بعد النشر وقبل تعليم الصفوف مرة واحدة:
docker compose run --rm -q app run 4 crash-relay
docker compose run --rm -q app statusوالمخرج سوف يكون كما يلي:
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، فلا شيء في النظام يعرف أنه موجود.
ولكي ترى جدول ال Outbox في خدمة الطلبات بعد التجربة:
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"والمخرج سوف يكون كما يلي:
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 أحداث للمسار نفسه.
حذف التجربة
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.
وفي مقالات هذا القسم ما يكمل هذا الموضوع، فجعل كل خطوة آمنة عند التكرار شرط لكل ما سبق، والتخزين المؤقت وحد المعدل يحميان الخدمات من الحمل:
دليلIdempotency: كيف تجعل العملية آمنة عند تكرارهالماذا تصل الرسالة نفسها أكثر من مرة، وكيف تكتب مستهلكاً يطبقها مرة واحدة فقط مهما تكررت.
دليلالتخزين المؤقت Caching بالتفصيل: الاستراتيجيات والإبطالكيف تختار نمط التخزين المؤقت المناسب، ومتى تبطل القيمة، وكيف تمنع انهيار قاعدة البيانات عندما ينتهي مفتاح مشهور.
دليلتحديد معدل الطلبات Rate Limiting وخوارزمياتهكيف تعمل خوارزميات تحديد معدل الطلبات، وأيها يناسب واجهتك، وكيف تطبقها على عدة سيرفرات وترد بالرمز 429 الصحيح.
دليلالطابور Queue أم بث الأحداث Event Streaming؟ما يحدث للرسالة بعد قراءتها هو الفرق بين طابور مثل RabbitMQ وسجل أحداث مثل Kafka، ومنه تعرف أيهما يناسب مشروعك.سجل التحديثات
- أكتوبر 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 بالتفصيل.
