Реализация Function Composition (Step Functions / Durable Functions)

Реализация Function Composition (Step Functions / Durable Functions)

Function Composition — это организация нескольких serverless функций в рабочий процесс с управлением состоянием, ветвлением, параллелизмом и обработкой ошибок. Один Lambda/Function вызывает другой напрямую — это антипаттерн: теряется история, сложно обрабатывать ошибки, нет видимости прогресса. Оркестраторы решают эти проблемы.

Когда нужна Function Composition

• Бизнес-процесс состоит из нескольких шагов с состоянием
• Нужны условия ветвления (if step_A succeeded, then step_B, else step_C)
• Параллельное выполнение нескольких функций с ожиданием результата
• Долгоживущие процессы (> 15 минут для Lambda)
• Человеческое подтверждение на каком-то шаге (wait for callback)

AWS Step Functions

State Machine Language (ASL) описывает рабочий процесс декларативно:

{
  "Comment": "Обработка заказа",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123:function:validate-order",
      "Next": "CheckInventory",
      "Retry": [{"ErrorEquals": ["Lambda.ServiceException"], "MaxAttempts": 3}],
      "Catch": [{
        "ErrorEquals": ["ValidationError"],
        "Next": "NotifyInvalidOrder"
      }]
    },
    "CheckInventory": {
      "Type": "Parallel",
      "Branches": [
        {"StartAt": "ReserveItems", "States": {"ReserveItems": {"Type": "Task", "Resource": "arn:...:reserve-items", "End": true}}},
        {"StartAt": "CalculateShipping", "States": {"CalculateShipping": {"Type": "Task", "Resource": "arn:...:calc-shipping", "End": true}}}
      ],
      "Next": "ProcessPayment"
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Parameters": {
        "FunctionName": "arn:...:process-payment",
        "Payload": {
          "taskToken.$": "$$.Task.Token",
          "orderId.$": "$.orderId"
        }
      },
      "Next": "FulfillOrder",
      "TimeoutSeconds": 300
    },
    "FulfillOrder": {"Type": "Task", "Resource": "arn:...:fulfill-order", "End": true},
    "NotifyInvalidOrder": {"Type": "Task", "Resource": "arn:...:notify-invalid", "End": true}
  }
}

.waitForTaskToken позволяет Step Functions ждать callback от внешней системы (платёжный шлюз) без polling. Платёжный шлюз вызывает SendTaskSuccess с токеном когда транзакция завершена.

Terraform для Step Functions

resource "aws_sfn_state_machine" "order_processing" {
  name     = "order-processing"
  role_arn = aws_iam_role.sfn_role.arn

  definition = templatefile("${path.module}/state_machine.json", {
    validate_lambda_arn = aws_lambda_function.validate_order.arn
    reserve_lambda_arn  = aws_lambda_function.reserve_items.arn
    payment_lambda_arn  = aws_lambda_function.process_payment.arn
    fulfill_lambda_arn  = aws_lambda_function.fulfill_order.arn
  })

  logging_configuration {
    log_destination        = "${aws_cloudwatch_log_group.sfn.arn}:*"
    include_execution_data = true
    level                  = "ERROR"
  }

  tracing_configuration {
    enabled = true  # X-Ray tracing
  }
}

Azure Durable Functions

.NET / Node.js / Python оркестратор на базе Azure Functions:

# orchestrator function
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    # Параллельное выполнение
    parallel_tasks = [
        context.call_activity("ReserveItems", context.get_input()),
        context.call_activity("CalculateShipping", context.get_input())
    ]
    results = yield context.task_all(parallel_tasks)
    
    # Ожидание внешнего события (человеческое подтверждение)
    approval = yield context.wait_for_external_event("ApprovalReceived")
    
    if approval:
        return (yield context.call_activity("FulfillOrder", context.get_input()))
    else:
        return (yield context.call_activity("CancelOrder", context.get_input()))

main = df.Orchestrator.create(orchestrator_function)

Durable Functions используют Azure Storage для хранения состояния. Оркестратор может ждать внешнего события неограниченно долго.

Обработка ошибок и компенсирующие транзакции

В распределённых процессах нет встроенных транзакций. Паттерн Saga — компенсирующие действия при откате:

"ProcessPayment": {
  "Type": "Task",
  "Resource": "...",
  "Catch": [{
    "ErrorEquals": ["PaymentFailed"],
    "Next": "CompensateReservation"
  }]
},
"CompensateReservation": {
  "Type": "Task",
  "Resource": "arn:...:release-reservation",
  "Next": "NotifyPaymentFailed"
}

Каждый шаг, который нужно откатить при ошибке, имеет компенсирующую функцию.

Видимость и мониторинг

Step Functions предоставляет visual workflow в Console — каждое выполнение можно просмотреть шаг за шагом. CloudWatch Metrics: ExecutionsStarted, ExecutionsSucceeded, ExecutionsFailed, ExecutionThrottled.

X-Ray интеграция даёт трейсинг через все Lambda функции в workflow.

Express vs Standard Workflows

Сроки реализации

• Проектирование state machine + ASL описание — 2-3 дня
• Lambda функции для каждого шага — 3-7 дней
• Step Functions state machine + IAM — 2-3 дня
• Обработка ошибок + компенсации — 2-3 дня
• Мониторинг + алерты + тестирование — 2-3 дня