Fluxos de trabalho

Os fluxos de trabalho dividem tarefas de longa duração em etapas distintas. Cada etapa persiste sua saída em um conjunto de estados compartilhado no banco de dados. Se PHP falhar durante uma etapa, o worker tenta executá-la novamente com todo o estado anterior intacto.

O que são fluxos de trabalho?

Um fluxo de trabalho é uma sequência ordenada de etapas. Cada etapa é uma classe PHP que recebe o estado acumulado de todas as etapas anteriores e retorna novos dados para mesclar ao estado. O fluxo de trabalho avança uma etapa por vez, com cada transição envolvida em uma transação atômica MySQL.

use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Acúmulo de estado

O conteúdo inicial que você passa para dispatch() se torna o estado inicial. As chaves retornadas por cada etapa são mescladas ao estado. As chaves que começam com _ são reservadas para valores de controle, como _next_step, e não são mescladas ao estado público:

dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

Cada etapa recebe o estado acumulado completo. Isso facilita a transferência de dados entre etapas sem armazenamento externo.

Capacidade de retomada

Os fluxos de trabalho são duráveis porque o estado é persistido no banco de dados após cada etapa. Se o worker falhar durante a etapa 1, o fluxo de trabalho reinicia na etapa 1 com a saída da etapa 0 já no estado.

O limite da etapa é uma única transação MySQL que, de forma atômica:

1. Atualiza o estado do fluxo de trabalho com a saída da etapa
2. Marca como concluído o trabalho da etapa atual
3. Enfileira o trabalho da próxima etapa

Isso garante que nenhum estado seja perdido, mesmo diante de falhas, timeouts ou erros fatais PHP.

Ciclo de vida do fluxo de trabalho

dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)

Gerenciamento de fluxos de trabalho

Verificar status:

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

Pausar e retomar:

Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

Tentar novamente um fluxo de trabalho com falha a partir da etapa que falhou:

Queuety::retry_workflow( $workflow_id );

Cancelar um fluxo de trabalho e executar manipuladores de limpeza:

Queuety::cancel_workflow( $workflow_id );

Consulte Cancelamento para saber mais sobre os manipuladores de limpeza e o que acontece quando um fluxo de trabalho é cancelado.

Opções do fluxo de trabalho

Defina a fila, a prioridade e o comportamento de novas tentativas para todas as etapas de um fluxo de trabalho:

use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

O método on_cancel() registra um manipulador de limpeza que é executado quando o fluxo de trabalho é cancelado. Consulte Cancelamento.

O método prune_state_after() permite a remoção automática das saídas antigas das etapas para manter limitado o estado do fluxo de trabalho. Consulte Limpeza do estado.

Use compensate_with() para anexar um manipulador de compensação por etapa. Use compensate_on_failure() quando quiser que as compensações das etapas concluídas sejam executadas automaticamente após uma falha do fluxo de trabalho, e não apenas em caso de cancelamento explícito. Consulte Compensação.

Use as proteções do fluxo de trabalho quando precisar de uma orquestração mais segura para tarefas de longa duração: version() adiciona uma versão no nível do aplicativo a uma definição, Queuety também armazena um hash determinístico da definição em cada execução, idempotency_key() torna o despacho durável diante de novas tentativas ou solicitações duplicadas, e max_transitions(), max_for_each_items() e max_state_bytes() interrompem execuções descontroladas antes que saiam do controle. Consulte Proteções.

Tipos de etapa

Queuety oferece onze tipos de etapa em um fluxo de trabalho:

• Cadeias sequenciais usando ->then() para execução uma etapa por vez
• Grupos paralelos usando ->parallel() para execução concorrente
• Repetição dinâmica para cada item usando ->for_each() para ramificações identificadas em tempo de execução e semântica explícita de conclusão
• Transferências assíncronas usando ->start_workflows() ou ->start_agents() para despachar fluxos de trabalho filhos independentes a partir de itens identificados em tempo de execução
• Ramificação condicional usando etapas nomeadas e _next_step
• Repetições usando ->repeat_until() e ->repeat_while() para criar retornos duráveis a etapas nomeadas anteriores
• Execução de fluxos de trabalho usando ->run_workflow() para composição aninhada
• Atrasos duráveis usando ->delay() para atrasos baseados em tempo
• Sinais usando ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() e ->wait_for_decision() para acionar eventos externos
• Dependências entre fluxos de trabalho usando ->wait_for_workflow(), ->wait_for_workflows() e ->wait_for_agents() para coordenação entre fluxos
• Etapas de transmissão usando a interface StreamingStep para saída durável baseada em blocos

Padrões comuns

Se estiver criando sistemas autônomos ou semiautônomos, comece por Orquestração de agentes. O conteúdo mostra como for_each(), start_agents(), wait_for_agents(), wait_for_workflow() e wait_for_decision() se combinam em fluxos reais de planejador/executor e revisão humana.