Protecciones para flujos de trabajo

Los flujos de trabajo al estilo de agentes son flexibles, pero también facilitan la creación de ejecuciones duplicadas o que un planificador genere más trabajo del previsto. Queuety ofrece protecciones ligeras en el nivel del flujo de trabajo para conservar esa flexibilidad sin perder el control.

Para conocer el modelo general de recursos que abarca la admisión de trabajadores y los metadatos de costes por trabajo, consulta Gestión de recursos.

Versionar una definición de flujo de trabajo

Usa version() para asignar una etiqueta de versión de la aplicación a una definición de flujo de trabajo:

Queuety::workflow( 'research_run' )
    ->version( 'research.v2' )
    ->then( PlanResearchStep::class )
    ->then( DraftResearchSummaryStep::class )
    ->dispatch();

La versión se almacena con el estado del flujo de trabajo, se muestra mediante workflow_status(), aparece en la salida de wp queuety workflow status y se incluye en las exportaciones del flujo de trabajo.

Queuety también almacena un definition_hash determinista para cada ejecución del flujo de trabajo. El hash se obtiene de la definición serializada del flujo y resulta útil para identificar exactamente qué estructura de flujo está ejecutando una tarea en curso después de una implementación.

Hacer idempotente el envío

Usa idempotency_key() cuando los clientes puedan reintentar la misma solicitud:

Queuety::workflow( 'research_run' )
    ->idempotency_key( 'brief:42:research' )
    ->then( PlanResearchStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

Si se vuelve a enviar un flujo de trabajo con una clave ya utilizada, Queuety devuelve el ID del flujo original en lugar de crear una ejecución duplicada. Las claves son globales para todos los nombres de flujos de trabajo, así que incluye el nombre del flujo en la clave cuando pueda haber varios flujos que compartan una.

Esto resulta especialmente útil para:

• Puntos de conexión API cuyos clientes o pasarelas pueden reintentar las solicitudes
• Consumidores de webhooks que pueden recibir entregas duplicadas
• Pasos de planificación que deben crear una sola ejecución de orquestación por entidad externa

Las claves de idempotencia se almacenan en la tabla queuety_workflow_dispatch_keys.

Establecer límites de uso de flujos de trabajo

Usa los límites de uso de flujos de trabajo para hacer fallar las ejecuciones que superen los límites previstos:

Queuety::workflow( 'research_run' )
    ->max_transitions( 20 )
    ->max_for_each_items( 12 )
    ->max_state_bytes( 32768 )
    ->max_cost_units( 40 )
    ->max_started_workflows( 8 )
    ->then( PlanResearchStep::class )
    ->for_each( 'tasks', ExecuteResearchTask::class, 'results' )
    ->dispatch();

max_transitions()

Limita cuántos pasos del flujo de trabajo pueden completarse antes de que falle la ejecución. Esto ayuda a proteger los flujos de trabajo que crean ramas dinámicamente o pueden volver a la misma sección mediante _next_step.

max_for_each_items()

Limita cuántos elementos descubiertos durante la ejecución puede expandir un solo paso for_each(). Resulta útil en sistemas de planificación y ejecución en los que un agente podría descubrir muchas más tareas de ramificación de las previstas.

max_state_bytes()

Limita el tamaño codificado del estado público del flujo de trabajo. Así se evita que los flujos de trabajo de larga duración crezcan hasta convertirse en grandes bloques de datos difíciles de inspeccionar, exportar o reintentar.

max_cost_units()

Limita las unidades de coste totales consumidas por los pasos completados. Las unidades de coste proceden de los metadatos del paso o del trabajo que Queuety resuelve al poner tareas en cola. El trabajo normal de usuario tiene el valor predeterminado 1, los elementos de orquestación como las demoras y las esperas usan 0, y los controladores pueden establecer un valor positivo mayor mediante config() o las propiedades del trabajo. Solo cost_units de una definición de paso puede establecer el valor 0.

Esto resulta útil cuando un flujo de trabajo puede volver a la misma sección o expandirse en tareas cuyo coste no se refleja solo en el recuento de pasos. Un planificador que repite una llamada costosa a un modelo podría mantenerse dentro de max_transitions() y aun así superar el límite de coste previsto.

max_started_workflows()

Limita cuántos flujos de trabajo de nivel superior puede iniciar una ejecución mediante start_workflows() o start_agents(). El límite se comprueba antes de enviar los flujos de trabajo secundarios, así que el flujo principal falla sin crear parcialmente un grupo que ya exceda el límite.

Cuando se supera un límite, Queuety hace fallar el flujo de trabajo inmediatamente en vez de reintentar el paso. Hay una excepción: si Queuety::signal() completa una espera y la transición resultante supera un límite, la excepción se lanza al llamador de signal(), se revierte la inserción de la señal y el flujo de trabajo sigue esperando.

Observar las protecciones

El estado del flujo de trabajo muestra:

• definition_version
• definition_hash
• idempotency_key
• budget

La carga útil de budget incluye los límites configurados y el uso actual, incluido el tamaño actual del estado público en bytes.

Ejemplo:

$state = Queuety::workflow_status( $workflow_id );

$state->definition_version; // 'research.v2'
$state->definition_hash;    // '4c7a...'
$state->idempotency_key;    // 'brief:42:research'
$state->budget;             // [ 'max_transitions' => 20, ... ]