Activity

Activity — это функции, выполняющие недетерминированные операции. Любую запись function.lua или process.lua можно зарегистрировать как Temporal activity, добавив метаданные.

Регистрация activity

Добавьте meta.temporal.activity для регистрации функции как activity:

- name: charge_payment
  kind: function.lua
  source: file://payment.lua
  method: charge
  modules:
    - http_client
    - json
  meta:
    temporal:
      activity:
        worker: app:worker

Поля метаданных

Реализация

Activity — это обычные функции Lua:

-- payment.lua
local http = require("http_client")
local json = require("json")

local function charge(input)
    local response, err = http.post("https://api.stripe.com/v1/charges", {
        headers = {
            ["Authorization"] = "Bearer " .. input.api_key,
            ["Content-Type"] = "application/json"
        },
        body = json.encode({
            amount = input.amount,
            currency = input.currency,
            source = input.token
        })
    })

    if err then
        return nil, err
    end

    return json.decode(response:body())
end

return { charge = charge }

Вызов activity

Из workflow используйте модуль funcs:

local funcs = require("funcs")

local result, err = funcs.call("app:charge_payment", {
    amount = 5000,
    currency = "usd",
    token = "tok_visa",
    api_key = ctx.stripe_key
})

if err then
    return nil, err
end

Параметры activity

Настройка тайм-аутов, повторных попыток и других параметров выполнения через построитель executor:

local funcs = require("funcs")

local executor = funcs.new():with_options({
    ["activity.start_to_close_timeout"] = "30s",
    ["activity.schedule_to_close_timeout"] = "5m",
    ["activity.heartbeat_timeout"] = "10s",
    ["activity.retry_policy"] = {
        maximum_attempts = 3,
        initial_interval = 1000,
        backoff_coefficient = 2.0,
        maximum_interval = 60000,
    }
})

local result, err = executor:call("app:charge_payment", input)

Executor неизменяем и может использоваться повторно. Создайте его один раз и используйте для нескольких вызовов:

local reliable = funcs.new():with_options({
    ["activity.start_to_close_timeout"] = "60s",
    ["activity.retry_policy"] = {
        maximum_attempts = 5,
        initial_interval = 2000,
        backoff_coefficient = 2.0,
        maximum_interval = 120000,
    }
})

local a, err = reliable:call("app:step_one", input)
local b, err = reliable:call("app:step_two", a)

Справочник параметров

Значения длительности принимают строки ("5s", "10m", "1h") или числа в миллисекундах.

Политика повторных попыток

Настройка автоматических повторных попыток для неуспешных activity:

["activity.retry_policy"] = {
    initial_interval = 1000,         -- миллисекунды до первой повторной попытки
    backoff_coefficient = 2.0,       -- множитель для каждой попытки
    maximum_interval = 300000,       -- максимальный интервал между попытками (мс)
    maximum_attempts = 10,           -- максимальное число попыток (0 = без ограничений)
    non_retryable_error_types = {    -- ошибки, для которых повтор не выполняется
        "INVALID",
        "PERMISSION_DENIED"
    }
}

Взаимосвязь тайм-аутов

|--- schedule_to_close_timeout --------------------------------|
|--- schedule_to_start_timeout ---|--- start_to_close_timeout -|
     (waiting in queue)                (executing)

• start_to_close_timeout: максимальное время выполнения самой activity. Наиболее часто используемый тайм-аут.
• schedule_to_close_timeout: общее время от момента планирования activity до её завершения, включая ожидание в очереди и повторные попытки.
• schedule_to_start_timeout: максимальное время ожидания activity в очереди задач до подхвата воркером.
• heartbeat_timeout: для долгоживущих activity — максимальное время между отчётами heartbeat.

Локальные activity

Локальные activity выполняются в процессе воркера workflow без отдельного опроса очереди задач:

- name: validate_input
  kind: function.lua
  source: file://validate.lua
  method: validate
  modules:
    - json
  meta:
    temporal:
      activity:
        worker: app:worker
        local: true

Сейчас local: true разбирается, но ведёт себя идентично обычной activity: она регистрируется и выполняется по стандартному пути activity. Отдельного выполнения локальной activity пока нет, поэтому этот флаг не меняет задержку, поведение очереди задач или heartbeat.

Именование activity

Activity регистрируются с полным ID записи в качестве имени:

namespace: app
entries:
  - name: charge_payment
    kind: function.lua
    # ...

Имя activity: app:charge_payment

Передача контекста

Значения контекста, установленные при запуске workflow, доступны внутри activity:

-- Spawner устанавливает контекст
local spawner = process.with_context({
    user_id = "user-1",
    tenant = "tenant-1",
})
local pid = spawner:spawn("app:order_workflow", "app:worker", order)

-- Activity читает контекст
local ctx = require("ctx")

local function process_order(input)
    local user_id = ctx.get("user_id")   -- "user-1"
    local tenant = ctx.get("tenant")     -- "tenant-1"
    -- используйте контекст для авторизации, логирования и т.д.
end

Activity, вызванные из workflow через funcs.new():with_context(), также передают контекст:

-- Внутри workflow
local executor = funcs.new():with_context({trace_id = "abc-123"})
local result, err = executor:call("app:charge_payment", input)

Контекст безопасности

Activity, запланированная в контексте безопасности, получает подписанный заголовок wippy-security, аудиторией которого является ID activity. Воркер проверяет подпись и аудиторию, затем сливает переданные значения ctx и полезную нагрузку безопасности в новый фрейм до запуска функции activity.

Это слияние происходит по принципу «всё или ничего» и фатально для activity при неудаче: activity возвращает ошибку до выполнения своего кода, поэтому она никогда не работает с частичным контекстом или непроверенным актором. Слияние не удаётся, если подпись или аудитория не проходят проверку, если конверт несогласован (актор без области или политики без актора) либо если названная в конверте политика не разрешается в локальном реестре безопасности — это и есть частая эксплуатационная причина: в развёртывании воркера отсутствует запись политики, которая была у вызывающей стороны.

Ключи подписи и проверки воркер берёт из записи temporal.client, на которую ссылается. См. Передача контекста безопасности.

Обработка ошибок

Возвращайте ошибки стандартным способом Lua:

local errors = require("errors")

local function charge(input)
    if not input.amount or input.amount <= 0 then
        return nil, errors.new({ kind = errors.INVALID, message = "amount must be positive" })
    end

    local response, err = http.post(url, options)
    if err then
        return nil, errors.wrap(err, "payment API failed")
    end

    if response:status() >= 400 then
        return nil, errors.new({ kind = errors.INVALID, message = "payment declined" })
    end

    return json.decode(response:body())
end

Объекты ошибок

Ошибки activity, переданные в workflow, содержат структурированные метаданные:

local result, err = funcs.call("app:charge_payment", input)
if err then
    err:kind()       -- error classification string
    err:retryable()  -- boolean, whether retry makes sense
    err:message()    -- human-readable error message
end

Режимы сбоя

Оба сбоя безопасности происходят во время слияния контекста, до запуска функции activity. Они не помечены как неповторяемые, поэтому политика повторных попыток activity продолжает перезапускать их; повторы не помогают, поскольку ни неверная подпись, ни отсутствующая запись политики не меняются между попытками. Ограничивайте maximum_attempts у activity, которые должны падать быстро, и читайте повторяющийся сбой Internal без вывода в лог activity как сбой слияния контекста, а не как неисправность самой activity.

local executor = funcs.new():with_options({
    ["activity.retry_policy"] = {maximum_attempts = 1}
})

local result, err = executor:call("app:missing_activity", input)
if err then
    print(err:kind())      -- "NotFound"
    print(err:retryable())  -- false
end

Процессы как activity

Записи process.lua тоже можно регистрировать как activity для долгоживущих операций:

- name: long_task
  kind: process.lua
  source: file://long_task.lua
  method: main
  modules:
    - http_client
  meta:
    temporal:
      activity:
        worker: app:worker

См. также

• Обзор — настройка
• Workflow — реализация workflow
• Функции — модуль функций
• Обработка ошибок — типы ошибок и паттерны