Flujos de Trabajo

Los flujos de trabajo persisten el estado de las operaciones de larga duración para que la ejecución pueda recuperarse después de fallos y reinicios. Son adecuados para procesos como pagos, cumplimiento de pedidos y aprobaciones de múltiples pasos.

Por qué Flujos de Trabajo

Las funciones conservan el estado en curso en memoria, mientras que los flujos de trabajo persisten el estado de ejecución:

Cómo Funcionan los Flujos de Trabajo

El código del flujo de trabajo parece código Lua regular:

local funcs = require("funcs")
local time = require("time")

local result, err = funcs.call("app.api:charge_card", payment)
if err then return nil, err end

time.sleep("24h")

local status, err = funcs.call("app.api:check_status", result.id)
if err then return nil, err end

if status == "failed" then
    local _, refund_err = funcs.call("app.api:refund", result.id)
    if refund_err then return nil, refund_err end
end

return status

El motor de flujos de trabajo intercepta las llamadas y registra sus resultados. Después de un fallo, reproduce la ejecución desde el historial registrado.

Dentro de un flujo de trabajo, cada destino de funcs.call() se ejecuta como una actividad de Temporal. Una entrada function.* de destino debe registrarse con un worker mediante meta.temporal.activity.worker; las entradas no registradas no están disponibles para el flujo de trabajo. Un destino de actividad process.* también necesita meta.options.default_host (o el campo heredado meta.default_host) para registrarse en el registro de funciones que usa el worker de Temporal. Consulte Actividades para ver un ejemplo de actividad de función y sus opciones.

Patrones de Flujo de Trabajo

Patrón Saga

Compensar en caso de fallo:

local funcs = require("funcs")

local inventory, err = funcs.call("app.inventory:reserve", items)
if err then return nil, err end

local payment, err = funcs.call("app.payments:charge", amount)
if err then
    local _, compensation_err = funcs.call("app.inventory:release", inventory.id)
    return nil, compensation_err or err
end

local shipping, err = funcs.call("app.shipping:create", order)
if err then
    local _, refund_err = funcs.call("app.payments:refund", payment.id)
    local _, release_err = funcs.call("app.inventory:release", inventory.id)
    return nil, refund_err or release_err or err
end

return {inventory = inventory, payment = payment, shipping = shipping}

Esperando Señales

Esperar eventos externos (decisiones de aprobación, webhooks, acciones de usuario):

local funcs = require("funcs")

local _, err = funcs.call("app.approvals:submit", request)
if err then return nil, err end

local inbox = process.inbox()
local msg, open = inbox:receive()  -- blocks until signal arrives
if not open then return nil, errors.new("workflow inbox closed") end

local decision, payload_err = msg:payload():data()
if payload_err then return nil, payload_err end

if decision.approved then
    return funcs.call("app.orders:fulfill", request.order_id)
else
    return funcs.call("app.notifications:send_rejection", request)
end

Elegir un modelo de cómputo :id=choosing-a-compute-model

Iniciando Flujos de Trabajo

Los flujos de trabajo se crean de la misma manera que los procesos: usando process.spawn() con un host diferente:

-- Spawn workflow on temporal worker
local pid, err = process.spawn("app.workflows:order_processor", "app:temporal_worker", order_data)
if err then return nil, err end

-- Send signals to workflow
local ok, err = process.send(pid, "update", {status = "approved"})
if err then return nil, err end
return ok

El llamador usa la misma API de spawn. El host determina si la entrada se ejecuta en un temporal.worker o en un process.host. El historial persistido y el replay se aplican solo en la ruta alojada por Temporal. Una entrada de flujo de trabajo ejecutada mediante un host de procesos normal tiene semántica de proceso en memoria y no obtiene la durabilidad de Temporal.

Fallo y Supervisión

Los procesos pueden ejecutarse como servicios supervisados usando process.service:

# Process definition
- name: session_handler
  kind: process.lua
  source: file://session_handler.lua
  method: main

# Supervised service wrapping the process
- name: session_manager
  kind: process.service
  process: app:session_handler
  host: app:processes
  lifecycle:
    auto_start: true
    restart:
      max_attempts: 10

Los flujos de trabajo no usan árboles de supervisión de procesos. El proveedor administra la persistencia y la recuperación; los reintentos de la aplicación siguen las políticas configuradas para el flujo de trabajo y sus actividades.

Configuración

Definición de proceso (creado dinámicamente):

- name: order_processor
  kind: workflow.lua
  source: file://order_processor.lua
  method: main
  meta:
    temporal:
      workflow:
        worker: app:temporal_worker
  modules:
    - funcs
    - time

Cada función o proceso invocado mediante funcs.call() también declara el worker de actividad. Por ejemplo:

- name: charge_card
  kind: function.lua
  source: file://charge_card.lua
  method: main
  meta:
    temporal:
      activity:
        worker: app:temporal_worker

Proveedor de flujos de trabajo:

- name: temporal_worker
  kind: temporal.worker
  client: app:temporal_client
  task_queue: "orders"
  lifecycle:
    auto_start: true

Consulte Temporal para infraestructura de flujos de trabajo en producción.

Véase también :id=see-also

• Funciones — Llamadas con ámbito de solicitud
• Modelo de procesos — Trabajo en segundo plano con estado
• Supervisión — Políticas de reinicio de procesos