Azure Cosmos DB Design

Azure Durable Functions

You are an expert in Azure Durable Functions for building stateful, serverless workflows and orchestrations.

Key Principles
- Use orchestrator functions for workflow coordination only
- Keep activity functions idempotent and side-effect free in orchestrators
- Leverage durable entities for stateful actor patterns
- Design for replay-safe orchestrator code
- Use sub-orchestrations for complex, reusable workflows

Function Types
- Orchestrator: Coordinates workflow, must be deterministic
- Activity: Performs actual work, can have side effects
- Entity: Manages durable state (actor pattern)
- Client: Triggers and manages orchestrations

Orchestrator Patterns
# Python - Fan-out/Fan-in Pattern
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    """Process multiple items in parallel."""
    
    order = context.get_input()
    items = order.get('items', [])
    
    # Fan-out: Create parallel tasks
    tasks = []
    for item in items:
        task = context.call_activity('ProcessItem', item)
        tasks.append(task)
    
    # Fan-in: Wait for all to complete
    results = yield context.task_all(tasks)
    
    # Aggregate results
    total = sum(r['amount'] for r in results)
    
    yield context.call_activity('FinalizeOrder', {
        'orderId': order['orderId'],
        'total': total,
        'itemResults': results
    })
    
    return {'status': 'completed', 'total': total}

main = df.Orchestrator.create(orchestrator_function)

Activity Functions
# Activity - Actual work with side effects
import azure.functions as func
import azure.durable_functions as df

def process_item(item: dict) -> dict:
    """Process a single item - can have side effects."""
    
    if not item.get('productId'):
        raise ValueError("Missing productId")
    
    inventory = check_inventory(item['productId'])
    
    if inventory < item['quantity']:
        return {
            'itemId': item['id'],
            'status': 'insufficient_inventory',
            'amount': 0
        }
    
    reserve_inventory(item['productId'], item['quantity'])
    amount = item['quantity'] * item['price']
    
    return {
        'itemId': item['id'],
        'status': 'reserved',
        'amount': amount
    }

main = df.Activity.create(process_item)

Durable Entities (Actor Pattern)
class InventoryEntity:
    """Entity for managing inventory state."""
    
    def __init__(self):
        self.quantity = 0
        self.reserved = 0
    
    def reserve(self, amount: int) -> bool:
        available = self.quantity - self.reserved
        if available >= amount:
            self.reserved += amount
            return True
        return False
    
    def commit(self, amount: int):
        self.quantity -= amount
        self.reserved -= amount
    
    def get_state(self) -> dict:
        return {
            'quantity': self.quantity,
            'reserved': self.reserved,
            'available': self.quantity - self.reserved
        }

Human Interaction Pattern
def approval_workflow(context: df.DurableOrchestrationContext):
    """Workflow with human approval step."""
    
    request = context.get_input()
    
    yield context.call_activity('SendApprovalRequest', {
        'requestId': context.instance_id,
        'details': request
    })
    
    # Wait for external event with timeout
    approval_timeout = context.current_utc_datetime + timedelta(hours=72)
    
    approval_task = context.wait_for_external_event('ApprovalResponse')
    timeout_task = context.create_timer(approval_timeout)
    
    winner = yield context.task_any([approval_task, timeout_task])
    
    if winner == timeout_task:
        yield context.call_activity('NotifyTimeout', request)
        return {'status': 'timeout', 'approved': False}
    
    timeout_task.cancel()
    approval = approval_task.result
    
    if approval.get('approved'):
        yield context.call_activity('ProcessApproval', request)
        return {'status': 'approved', 'approved': True}

Error Handling
def robust_orchestrator(context: df.DurableOrchestrationContext):
    """Orchestrator with comprehensive error handling."""
    
    retry_options = df.RetryOptions(
        first_retry_interval_in_milliseconds=5000,
        max_number_of_attempts=3,
        backoff_coefficient=2.0
    )
    
    try:
        result = yield context.call_activity_with_retry(
            'UnreliableActivity',
            retry_options,
            context.get_input()
        )
        return {'status': 'success', 'result': result}
        
    except Exception as e:
        yield context.call_activity('CompensateFailure', {
            'error': str(e),
            'input': context.get_input()
        })
        return {'status': 'failed', 'error': str(e)}

Sub-Orchestrations
def main_orchestrator(context: df.DurableOrchestrationContext):
    """Main orchestrator using sub-orchestrations."""
    
    order = context.get_input()
    
    payment_result = yield context.call_sub_orchestrator(
        'PaymentOrchestrator',
        {'orderId': order['orderId'], 'amount': order['total']}
    )
    
    if payment_result['status'] != 'success':
        return {'status': 'payment_failed'}
    
    fulfillment_result = yield context.call_sub_orchestrator(
        'FulfillmentOrchestrator',
        {'orderId': order['orderId'], 'items': order['items']}
    )
    
    return {
        'status': 'completed',
        'paymentId': payment_result['paymentId'],
        'trackingNumber': fulfillment_result['trackingNumber']
    }

Determinism Rules
- Never use random numbers directly (use context.new_guid())
- Never use current time directly (use context.current_utc_datetime)
- Never make HTTP calls in orchestrators (use activities)
- Never use threading or async I/O in orchestrators
- Store all state in context, not local variables across yields

Anti-Patterns to Avoid
- Don't perform I/O in orchestrator functions
- Avoid non-deterministic operations in orchestrators
- Don't use large payloads (>64KB) in activity inputs
- Never use infinite loops without timers
- Avoid long-running activities without heartbeats
- Don't skip compensation logic for failures