Workers e Tarefas Externas

Duas classes de trabalho assíncrono coexistem no motor, com mecanismos de retomada completamente diferentes:

• Tarefas internas pendentes (ExecutableTask com status: PENDING e dueDate no passado) — geradas por nós commitBefore: true, timers e retries agendados. São recuperadas automaticamente pelo próprio motor via polling interno (TaskAcquirer).
• Tarefas externas (ExternalTask) — geradas por nós EXTERNAL_TASK. Não há polling: a conclusão é sempre uma chamada explícita e externa a completeExternalTask(...).

TaskAcquirer: o coração do processamento assíncrono interno

io.kikwiflow.execution.TaskAcquirer implementa Runnable e roda em um único virtual thread dedicado (Thread.ofVirtual().name("kikwiflow-acquirer-", 0)), iniciado por KikwiflowEngine.start() — que, por sua vez, é chamado automaticamente pelo Spring via @Bean(initMethod = "start", destroyMethod = "stop") no KikwiflowAutoConfiguration. Não há nenhuma configuração adicional necessária para ligar o processamento assíncrono: subir a aplicação Spring Boot já é suficiente.

@Override
public void run() {
    while (running) {
        int availablePermits = concurrencyLimit.availablePermits();
        if (availablePermits <= 0) { Thread.sleep(taskAcquisitionIntervalMillis); continue; }

        int limitToFetch = Math.min(availablePermits, taskAcquisitionMaxTasks);
        List<ExecutableTask> taskList = kikwiEngineRepository.findAndLockDueTasks(
                Instant.now(), limitToFetch, workerId, lockTimeoutMillis);

        for (ExecutableTask task : taskList) {
            concurrencyLimit.acquireUninterruptibly();
            workerExecutor.submit(() -> {
                try { engine.executeFromTask(task); }
                finally { concurrencyLimit.release(); }
            });
        }
        Thread.sleep(taskAcquisitionIntervalMillis);
    }
}

Cada tarefa adquirida é despachada para um pool de virtual threads (Executors.newVirtualThreadPerTaskExecutor()) — o Kikwiflow assume Java 21+ e usa threads virtuais de propósito para permitir alta concorrência de I/O (chamadas de TaskHandler que fazem rede) sem o custo de memória de milhares de threads de plataforma.

Controle de concorrência: Semaphore

O número de tarefas em execução simultânea por instância do motor é limitado por um Semaphore inicializado com kikwiflow.execution.max-concurrent-tasks. O poller só busca no banco a quantidade de permits disponíveis (min(availablePermits, taskAcquisitionMaxTasks)) — isso evita que uma instância adquira mais trabalho do que consegue processar, deixando tarefas travadas em LOCKED desnecessariamente.

findAndLockDueTasks: locking otimista distribuído

findAndLockDueTasks(now, limit, workerId, lockTimeoutMillis) é o único ponto de contato entre o poller e o repositório de persistência — ele busca tarefas com dueDate <= now e as marca atomicamente como adquiridas por este workerId (um identificador único por instância: <instanceName>-<uuid-curto>), com um TTL de lock (lockTimeoutMillis). Isso é o que permite múltiplas instâncias do Kikwiflow rodarem em paralelo (escala horizontal) sem duas instâncias processarem a mesma tarefa simultaneamente — o mecanismo concreto de locking (otimista via campo de versão, ou findOneAndUpdate atômico) é responsabilidade da implementação de KikwiEngineRepository (ver Persistência e Consistência Transacional).

:::warning lockTimeoutMillis precisa ser realista Se um worker trava (crash do processo, OOM) enquanto segura o lock de uma tarefa, essa tarefa só volta a ficar elegível para outro worker depois que lockTimeoutMillis expira. Um valor baixo libera tarefas órfãs mais rápido, mas arrisca duas instâncias processarem a mesma tarefa se uma delas apenas está lenta (não morta). O application.yml do sample-onboarding-process usa 1200 (1.2s) — bastante agressivo, adequado para demonstração, não necessariamente para produção com TaskHandlers de I/O mais lento. :::

Graceful shutdown

KikwiflowEngine.stop() → TaskAcquirer.stop() interrompe o poller e aguarda até kikwiflow.execution.shutdown-grace-period-seconds pela conclusão de tarefas em voo antes de forçar shutdownNow() em ambos os executors (poller e workers). Isso é automaticamente ligado ao ciclo de vida do Spring (destroyMethod = "stop"), então um SIGTERM padrão (Kubernetes, systemd) já aciona o shutdown gracioso sem configuração adicional.

TaskExecutor: o despachante

io.kikwiflow.execution.TaskExecutor é deliberadamente minúsculo — sua única responsabilidade é, dado um ExecutionContext, verificar se o nó atual é uma ExecutableTaskDefinition com executor preenchido, resolver o bean TaskHandler correspondente via TaskHandlerResolver, e invocar handle(executionContext):

public void execute(ExecutionContext executionContext){
    FlowNodeDefinition node = executionContext.getFlowNode();
    if (node instanceof ExecutableTaskDefinition serviceTask) {
        if (serviceTask.executor() != null) {
            TaskHandler taskHandler = taskHandlerResolver.resolve(serviceTask.executor())
                    .orElseThrow(() -> new BadDefinitionExecutionException("TaskHandler not found: " + serviceTask.executor()));
            taskHandler.handle(executionContext);
        } else {
            throw new BadDefinitionExecutionException("Invalid execution method for task " + serviceTask.id());
        }
    }
}

Note que essa resolução de bean acontece em toda execução, não apenas no deploy — o DeployValidator já garante na hora do deploy que o bean existe (ver Navegação e Gateways de Decisão), mas a resolução em runtime permanece dinâmica, permitindo, por exemplo, trocar a implementação de um TaskHandler via @Primary/perfis Spring sem precisar reimplantar o processo.

ExternalTask: ciclo de vida completo

Um EXTERNAL_TASK gera um registro ExternalTask (não um ExecutableTask) — sem dueDate, sem retry policy, sem execução automática. Ele fica em ExternalTaskStatus.CREATED até uma das seguintes chamadas via KikwiflowEngine, todas expostas por kikwi-management-rest:

Método	Endpoint REST	Efeito
claim(externalTaskId, assignee, identityContext)	(via ExternalTaskOperationsRestApi)	Marca a tarefa como reivindicada por um assignee — útil para filas de trabalho humano, evitando que dois operadores peguem a mesma tarefa.
unclaim(externalTaskId)	idem	Libera a reivindicação.
completeExternalTask(externalTaskId, variables, identityContext)	PUT /external-tasks/{id}/complete (via ExternalTaskCommandController)	Injeta variáveis no processo e retoma a execução síncrona a partir do próximo nó, na mesma chamada.

@Override
public ProcessInstance completeExternalTask(String id, CompleteExternalTaskRequest req, IdentityContext identityContext) {
    return engine.completeExternalTask(id, req.variables(), identityContext);
}

completeExternalTask valida que o tenantId da tarefa bate com o do IdentityContext (lançando SecurityException em caso de divergência — ver Segurança e Multi-tenancy), mescla as variáveis fornecidas na instância, e chama Navigator.determineNextContinuation seguido de ProcessExecutionManager.executeFlow — ou seja, do ponto de vista do motor, completar uma tarefa externa é estruturalmente idêntico a retomar de um commitBefore: a execução volta a rodar em memória, síncrona, até a próxima fronteira.

Consultando tarefas externas pendentes

O lado de leitura (quais tarefas externas estão abertas, para preencher uma inbox de trabalho humano ou um dashboard de workers) é responsabilidade de kikwi-runtime-query-api/ExternalTaskQueryService, exposto via ExternalTaskQueryController — deliberadamente separado do caminho de comando, seguindo o princípio de CQRS descrito na Visão Geral.

Boundary events em tarefas externas

Tarefas externas frequentemente têm timers de borda anexados (ex.: expirar após N minutos sem resposta — ver Timers e Agendamento). Quando o boundary timer interrompe a ExternalTask, o ContinuationService remove tanto a ExternalTask original quanto a ExecutableTask do timer, e emite um evento FLOW_NODE_FINISHED com nodeExecutionStatus: INTERRUPTED referenciando o nó interrompido — visível em Eventos e Observabilidade.