إتقان Temporal: بناء سير عمل مرنة وطويلة الأمد

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

القيمة الجوهرية الأساسية لـ Temporal هي "التنفيذ الدائم" (Durable Execution). على عكس الأكواد التقليدية التي تعمل حتى تتعطل أو تكتمل، يسجل Temporal تاريخ كل استدعاء دالة. إذا تعطلت العملية أثناء التنفيذ، يمكن لـ Temporal استعادة العملية من نقطة التفتيش الأخيرة ومواصلة التنفيذ من حيث توقفت بالضبط. هذا التحول في النموذج ينقل تعقيد إدارة الحالة من المطور إلى المنصة.

البنية الأساسية: سير العمل والأنشطة

لفهم Temporal، يجب أن تميز بين مكوينيه الرئيسيين: سير العمل (Workflows) والأنشطة (Activities).

• سير العمل: منطق التنسيق. فهي تحدد ترتيب العمليات، وتتعامل مع إعادة المحاولة، وتدير فترات انتهاء الصلاحية. من الضروري أن يكون كود سير العمل حتميًّا (deterministic). لا يمكنك استخدام مولدات الأرقام العشوائية، أو الطوابع الزمنية الحالية، أو استدعاءات قاعدة البيانات المباشرة داخل دالة سير العمل مباشرة. بدلاً من ذلك، تتفاعل مع العالم الخارجي عبر الأنشطة.
• الأنشطة: العمل الفعلي الذي له آثار جانبية. إرسال البريد الإلكتروني، أو الاستعلام عن قاعدة البيانات، أو استدعاء واجهة برمجة تطبيقات خارجية يحدث هنا. الأنشطة غير حتمية وتُنفذ بواسطة عمال Temporal. إذا فشل نشاط، يمكن لـ Temporal إعادة المحاولة تلقائيًا بناءً على السياسات المهيأة دون إعادة تنفيذ سير العمل بأكمله من البداية.

التطبيق العملي في Python

لنلقِ نظرة على مثال عملي لسير عمل معالجة المدفوعات باستخدام SDK الخاص بـ Python من Temporal. يضمن هذا النمط بقاء النظام متسقًا حتى إذا أعيد تشغيل الخادم بين تحصيل البطاقة وتحديث سجل المستخدم.

import time
from temporalio.client import Client
from temporalio.common import RetryPolicy
from temporalio.contrib.opentelemetry import TracingWorkflowInterceptor
from temporalio.worker import Worker
from temporalio.workflow import (
    ActivityError,
    define_workflow,
    workflow_activity,
    workflow_run,
    workflow_signal,
)

# 1. Define the Activities (The "Side Effects")
@workflow_activity(name="charge_credit_card")
async def charge_credit_card(amount: float, card_id: str) -> str:
    """Simulate a credit card charge."""
    # In a real app, this calls a payment gateway
    if amount > 1000:
        raise Exception("Payment declined: Limit exceeded")
    return f"tx_{int(time.time())}_{card_id}"

@workflow_activity(name="update_user_balance")
async def update_user_balance(user_id: str, transaction_id: str) -> None:
    """Simulate updating the user's internal ledger."""
    # In a real app, this writes to the database
    print(f"Updating balance for {user_id} with tx {transaction_id}")

# 2. Define the Workflow (The "Orchestration")
@define_workflow(name="PaymentProcessingWorkflow")
class PaymentProcessingWorkflow:
    async def __init__(self, amount: float, card_id: str, user_id: str):
        self.amount = amount
        self.card_id = card_id
        self.user_id = user_id

    async def run(self):
        """
        This code is deterministic. 
        Temporal will replay this logic based on the event history.
        """
        # Execute the activity. 
        # If this fails, Temporal retries it according to the default policy.
        transaction_id = await self._activity(
            charge_credit_card, 
            self.amount, 
            self.card_id
        )
        
        # Wait for a signal (e.g., manual approval) if needed
        # await self._signal("approval_received")

        # Update the ledger
        await self._activity(
            update_user_balance, 
            self.user_id, 
            transaction_id
        )
        
        return transaction_id

    # Helper to call activities
    async def _activity(self, activity_func, *args, **kwargs):
        return await activity_func(*args, **kwargs)

# 3. Run a Worker (This is what your application service runs)
async def main():
    client = await Client.connect("localhost:7233", namespace="default")
    
    async with Worker(
        client,
        task_queue="payment-queue",
        workflows=[PaymentProcessingWorkflow],
        activities=[
            charge_credit_card,
            update_user_balance
        ]
    ) as worker:
        print("Worker ready")
        await worker.run()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

لماذا تختار Temporal بدلاً من البدائل؟

بينما تقدم أنظمة مثل AWS Step Functions أو Azure Durable Functions قدرات مشابهة، فإنها غالبًا ما تقيدك بمزودي سحابة محددين أو تجبرك على تعريف سير العمل عبر JSON/YAML. Temporal مفتوح المصدر وقابل للاستضافة الذاتية، مما يمنحك تحكمًا كاملًا في بنيتك التحتية. علاوة على ذلك، لأنك تكتب سير العمل بلغات شائعة (Python، Go، TypeScript، Java، .NET)، يمكنك الاستفادة من أكوادك الحالية، وأطر عمل الاختبارات الوحدوية، ودعم أدوات تطوير البرمجيات (IDE).

إحدى أقوى الميزات هي الرؤية (Visibility). مع Temporal، لا تحتاج إلى بناء لوحة سجلات مخصصة لرؤية حالة سير عملك. يتيح لك واجهة Temporal UI فحص أي تنفيذ لسير العمل، ورؤية التسلسل الدقيق للأحداث، وعرض الكود الذي تم تنفيذه، وحتى إطلاق إشارات لتغيير سلوك سير العمل الجارية في الوقت الفعلي.

الخاتمة

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