Activities

Las activities de Temporal ejecutan operaciones no deterministas. Registre una entrada function.lua o process.lua como activity mediante sus metadatos.

Los fragmentos son recetas de API. El ejemplo de pago es ilustrativo y requiere una entrada de entorno propiedad de la aplicación, permiso env.get para la credencial, permiso http_client.request para la URL del proveedor y un contrato con el proveedor de pagos.

Registrar Activities

Agregue meta.temporal.activity para registrar una función como activity:

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

Campos de Metadatos

Implementación

Las activities son funciones Lua normales. Mantenga las credenciales fuera de las entradas del workflow, porque Temporal las conserva en su historial. Este ejemplo lee la clave de pago del registro de entorno dentro de la activity. El proveedor de ejemplo acepta una solicitud de cobro JSON y devuelve una respuesta JSON. La correspondencia de estados es una política propiedad de la aplicación: sustituya la URL, los campos de solicitud y respuesta y la correspondencia de fallos por el contrato de su proveedor.

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

local function payment_error(status)
    if status == 408 then
        return errors.new({kind = errors.TIMEOUT, message = "payment provider timed out", retryable = true})
    elseif status == 429 then
        return errors.new({kind = errors.RATE_LIMITED, message = "payment provider rate limited the request", retryable = true})
    elseif status >= 500 then
        return errors.new({kind = errors.UNAVAILABLE, message = "payment provider is unavailable", retryable = true})
    end
    return errors.new({kind = errors.INVALID, message = "payment request was rejected", retryable = false})
end

local function charge(input)
    local api_key, env_err = env.get("PAYMENTS_API_KEY")
    if env_err then return nil, env_err end

    local body, encode_err = json.encode({
        amount = input.amount,
        currency = input.currency,
        payment_token = input.payment_token
    })
    if encode_err then
        return nil, encode_err
    end

    local response, err = http.post("https://payments.example.com/v1/charges", {
        headers = {
            ["Authorization"] = "Bearer " .. api_key,
            ["Content-Type"] = "application/json"
        },
        body = body
    })

    if err then
        return nil, err
    end

    if response.status_code >= 400 then
        return nil, payment_error(response.status_code)
    end

    return json.decode(response.body)
end

return { charge = charge }

Llamar Activities

Desde workflows, use el módulo funcs:

local funcs = require("funcs")

local result, err = funcs.call("app:charge_payment", {
    amount = 5000,
    currency = "usd",
    payment_token = "payment-token-123"
})

if err then
    return nil, err
end

Opciones de Activity

Configure timeouts, comportamiento de reintentos y otros parámetros de ejecución usando el constructor de 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)

El executor es inmutable y reutilizable. Constrúyalo una vez y úselo para múltiples llamadas:

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)
if err then
    return nil, err
end
local b, err = reliable:call("app:step_two", a)
if err then
    return nil, err
end

Referencia de Opciones

Los valores de duración aceptan cadenas ("5s", "10m", "1h") o milisegundos como números.

Use los nombres canónicos activity.* en código nuevo. Los alias heredados temporal.activity.* siguen aceptándose por compatibilidad.

local executor = funcs.new():with_options({
    ["activity.summary"] = "Charge the order payment",
    ["activity.priority"] = {
        priority_key = 10,
        fairness_key = "customer-123",
        fairness_weight = 1.0,
    },
    ["activity.name"] = "charge-payment",
    ["activity.versioning_intent"] = "use_assignment_rules",
})

Política de Reintentos

Configurar comportamiento automático de reintentos para activities fallidas:

["activity.retry_policy"] = {
    initial_interval = 1000,         -- ms before first retry
    backoff_coefficient = 2.0,       -- multiplier for each retry
    maximum_interval = 300000,       -- max interval between retries (ms)
    maximum_attempts = 10,           -- max retry attempts (0 = unlimited)
    non_retryable_error_types = {    -- errors that skip retries
        "Invalid",
        "PermissionDenied"
    }
}

Relaciones entre Timeouts

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

• start_to_close_timeout: Cuánto tiempo puede ejecutarse la activity. Es el timeout más comúnmente usado.
• schedule_to_close_timeout: Tiempo total desde que la activity se programa hasta que se completa, incluyendo tiempo de espera en cola y reintentos.
• schedule_to_start_timeout: Tiempo máximo que la activity puede esperar en la cola de tareas antes de que un worker la tome.
• heartbeat_timeout: Para activities de larga ejecución, el tiempo máximo entre reportes de heartbeat.

Activities Locales

El campo local se acepta en una activity:

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

Actualmente local: true se parsea pero se comporta igual que una activity regular: se registra y se ejecuta a través de la ruta estándar de activities. Todavía no existe una ejecución diferenciada de activity local, por lo que no cambia la latencia, el comportamiento de la cola de tareas ni el heartbeating.

Nombrado de Activities

Las activities se registran con su ID de entrada completo como nombre:

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

Nombre de activity: app:charge_payment

Propagación de Contexto

Los valores de contexto establecidos al hacer spawn del workflow están disponibles dentro de las activities:

-- Spawner sets context
local spawner = process.with_context({
    user_id = "user-1",
    tenant = "tenant-1",
})
local pid, err = spawner:spawn("app:order_workflow", "app:worker", order)
if err then
    return nil, err
end

-- Activity reads context
local ctx = require("ctx")

local function process_order(input)
    local user_id, user_err = ctx.get("user_id")   -- "user-1"
    if user_err then return nil, user_err end
    local tenant, tenant_err = ctx.get("tenant")   -- "tenant-1"
    if tenant_err then return nil, tenant_err end
    -- use context for authorization, logging, etc.
end

Las activities llamadas desde un workflow con funcs.new():with_context() también propagan contexto:

-- Inside workflow
local executor = funcs.new():with_context({trace_id = "abc-123"})
local result, err = executor:call("app:charge_payment", input)

Contexto de Seguridad

Una activity programada bajo un contexto de seguridad recibe la cabecera firmada wippy-security, con audiencia dirigida al ID de la activity. El worker verifica la firma y la audiencia, luego fusiona los valores de ctx propagados y el payload de seguridad sobre un frame nuevo antes de que se ejecute la función de la activity.

Esa fusión es de todo o nada y fatal para la activity si falla: la activity retorna un error antes de que su código se ejecute, por lo que nunca se ejecuta con contexto parcial o con un actor no verificado. Una fusión falla cuando la firma o la audiencia no se verifican, cuando el envelope es inconsistente (un actor sin scope, o políticas sin actor), o cuando una política nombrada en el envelope no se resuelve en el registro de seguridad local — que es la causa operativa común: al despliegue del worker le falta una entrada de política que el llamador sí tenía.

El worker toma sus claves de firma y verificación de la entrada temporal.client a la que hace referencia. Consulte Propagación del contexto de seguridad.

Manejo de Errores

Retorne errores mediante el patrón estándar de Lua:

local errors = require("errors")

-- Replace this mapping with the payment provider's documented error contract.
local function payment_error(status)
    if status == 408 then
        return errors.new({kind = errors.TIMEOUT, message = "payment provider timed out", retryable = true})
    elseif status == 429 then
        return errors.new({kind = errors.RATE_LIMITED, message = "payment provider rate limited the request", retryable = true})
    elseif status >= 500 then
        return errors.new({kind = errors.UNAVAILABLE, message = "payment provider is unavailable", retryable = true})
    end
    return errors.new({kind = errors.INVALID, message = "payment request was rejected", retryable = false})
end

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

Objetos de Error

Los errores de activity propagados a workflows portan metadatos estructurados:

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

Modos de Fallo

Ambos fallos de seguridad ocurren durante la fusión del contexto, antes de que se ejecute la función de la activity. No están marcados como no reintentables, por lo que la política de reintentos de la activity sigue reintentándolos; los reintentos no ayudan, porque ni una firma inválida ni una entrada de política faltante cambian entre intentos. Limite maximum_attempts en las activities que quiera que fallen rápido, y lea un fallo Internal repetido sin salida en el log de la activity como un fallo de fusión de contexto en lugar de un defecto en la 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

Activities de Proceso

Las entradas process.lua también pueden registrarse como activities para operaciones de larga ejecución:

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

Ver También

• Overview - Configuración
• Workflows - Implementación de workflows
• Funciones - Módulo de funciones
• Manejo de errores - Tipos de error y patrones