Azure Durable Functions

# 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
# Python - Fan-out/Fan-in Pattern
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    """Process multiple items in parallel."""
    
    # Get input
    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)
    
    # Continue workflow
    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
```python
# 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."""
    
    # Validate item
    if not item.get('productId'):
        raise ValueError("Missing productId")
    
    # Check inventory (external call)
    inventory = check_inventory(item['productId'])
    
    if inventory < item['quantity']:
        return {
            'itemId': item['id'],
            'status': 'insufficient_inventory',
            'amount': 0
        }
    
    # Reserve inventory
    reserve_inventory(item['productId'], item['quantity'])
    
    # Calculate amount
    amount = item['quantity'] * item['price']
    
    return {
        'itemId': item['id'],
        'status': 'reserved',
        'amount': amount
    }

main = df.Activity.create(process_item)
```

## Durable Entities (Actor Pattern)
```python
# Entity - Stateful actor
import azure.durable_functions as df
import json

class InventoryEntity:
    """Entity for managing inventory state."""
    
    def __init__(self):
        self.quantity = 0
        self.reserved = 0
    
    def add(self, amount: int):
        self.quantity += amount
    
    def reserve(self, amount: int) -> bool:
        available = self.quantity - self.reserved
        if available >= amount:
            self.reserved += amount
            return True
        return False
    
    def release(self, amount: int):
        self.reserved = max(0, self.reserved - amount)
    
    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
        }

# Entity function
def entity_function(context: df.DurableEntityContext):
    entity = InventoryEntity()
    
    # Load existing state
    state = context.get_state(lambda: entity.__dict__)
    entity.__dict__.update(state)
    
    # Handle operation
    operation = context.operation_name
    input_data = context.get_input()
    
    if operation == "add":
        entity.add(input_data)
    elif operation == "reserve":
        result = entity.reserve(input_data)
        context.set_result(result)
    elif operation == "release":
        entity.release(input_data)
    elif operation == "commit":
        entity.commit(input_data)
    elif operation == "get":
        context.set_result(entity.get_state())
    
    # Save state
    context.set_state(entity.__dict__)

main = df.Entity.create(entity_function)
```

## Human Interaction Pattern
```python
def approval_workflow(context: df.DurableOrchestrationContext):
    """Workflow with human approval step."""
    
    request = context.get_input()
    
    # Send approval request
    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:
        # Timeout - auto-reject
        yield context.call_activity('NotifyTimeout', request)
        return {'status': 'timeout', 'approved': False}
    
    # Cancel timer if approval received
    timeout_task.cancel()
    
    approval = approval_task.result
    
    if approval.get('approved'):
        yield context.call_activity('ProcessApproval', request)
        return {'status': 'approved', 'approved': True}
    else:
        yield context.call_activity('ProcessRejection', request)
        return {'status': 'rejected', 'approved': False}

main = df.Orchestrator.create(approval_workflow)
```

## Client Operations
```python
import azure.functions as func
import azure.durable_functions as df

async def http_start(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    """HTTP trigger to start orchestration."""
    
    client = df.DurableOrchestrationClient(starter)
    
    # Parse request
    request_body = req.get_json()
    
    # Start orchestration
    instance_id = await client.start_new(
        'OrderOrchestrator',
        client_input=request_body,
        instance_id=request_body.get('orderId')  # Optional custom ID
    )
    
    # Return management URLs
    return client.create_check_status_response(req, instance_id)

async def raise_event(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    """Raise external event to running orchestration."""
    
    client = df.DurableOrchestrationClient(starter)
    
    instance_id = req.route_params.get('instanceId')
    event_name = req.route_params.get('eventName')
    event_data = req.get_json()
    
    await client.raise_event(instance_id, event_name, event_data)
    
    return func.HttpResponse(status_code=202)
```

## Error Handling
```python
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,
        max_retry_interval_in_milliseconds=60000
    )
    
    try:
        # Activity with retry
        result = yield context.call_activity_with_retry(
            'UnreliableActivity',
            retry_options,
            context.get_input()
        )
        
        return {'status': 'success', 'result': result}
        
    except Exception as e:
        # Compensation logic
        yield context.call_activity('CompensateFailure', {
            'error': str(e),
            'input': context.get_input()
        })
        
        return {'status': 'failed', 'error': str(e)}

main = df.Orchestrator.create(robust_orchestrator)
```

## Sub-Orchestrations
```python
def main_orchestrator(context: df.DurableOrchestrationContext):
    """Main orchestrator using sub-orchestrations."""
    
    order = context.get_input()
    
    # Call sub-orchestration for payment
    payment_result = yield context.call_sub_orchestrator(
        'PaymentOrchestrator',
        {'orderId': order['orderId'], 'amount': order['total']}
    )
    
    if payment_result['status'] != 'success':
        return {'status': 'payment_failed'}
    
    # Call sub-orchestration for fulfillment
    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']
    }

main = df.Orchestrator.create(main_orchestrator)
```

## 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