Persistência e Consistência Transacional

Toda mutação de estado no Kikwiflow — criar uma instância, mover um token de execução, abrir um incidente, resolver um ramo paralelo — passa por um único ponto de entrada: KikwiEngineRepository.commitWork(UnitOfWork). Este documento descreve o contrato de persistência do motor e como ele garante atomicidade mesmo em operações que tocam múltiplas coleções/tabelas simultaneamente.

KikwiEngineRepository: o contrato CQRS completo

public interface KikwiEngineRepository extends QueryRepository, CommandRepository {
    void ensureIndexes();
}

• CommandRepository — mutações: saveProcessDefinition, commitWork, findAndLockDueTasks, addVariables, claim/unclaim, deleteProcessInstanceById.
• QueryRepository — leituras: consultas pontuais por ID, contadores (countOpenIncidentsByProcessDefinition, etc.), métricas agregadas (getProcessMacroMetrics, getMetricsByNodeForProcessDefinition), e os builders fluentes createProcessInstanceQuery()/createExternalTaskQuery() usados pela API de busca avançada.

Qualquer implementação alternativa de persistência (fora do Mongo) precisa implementar essa interface por completo — hoje as duas implementações disponíveis são o addon MongoDB (kikwi-runtime-persistence-mongodb, produção) e o addon in-memory (kikwi-in-memory-addons, testes/demos).

UnitOfWork: a unidade atômica de mudança de estado

public record UnitOfWork(
        ProcessInstance instanceToCreate,
        ProcessInstance instanceToUpdate,
        ProcessInstance instanceToDelete,
        List<ExecutableTask> executableTasksToCreate,
        List<ExternalTask> externalTasksToCreate,
        List<String> executableTasksToDelete,
        List<ExecutableTask> executableTasksToUpdate,
        List<String> externalTasksToDelete,
        List<OutboxEventEntity> events,
        List<Incident> incidentsToCreate,
        List<Incident> incidentsToUpdate,
        List<String> incidentsToResolve,
        List<String> finishedNodeDefinitions,
        List<BranchPullIntention> branchPullIntentions,
        Map<String, VariableOperation> variableOperations
        ) {}

Cada campo é opcional (pode ser null/vazio) — um UnitOfWork é montado apenas com o subconjunto de mudanças relevantes para aquela operação específica. Por exemplo, FailureHandler.handleFailure monta um UnitOfWork contendo apenas executableTasksToUpdate e incidentsToCreate, deixando todo o resto null:

UnitOfWork uow = new UnitOfWork(
        null, null, null, null, null, null,
        tasksToUpdate, null, null, incidentsToCreate,
        null, null, null, null, null);
repository.commitWork(uow);

Esse desenho — um único record grande e "achatado" em vez de múltiplos métodos de escrita separados — é o que permite ao ContinuationService (o consumidor mais complexo do UnitOfWork) expressar em uma única chamada transacional operações que, de outra forma, exigiriam múltiplas idas ao banco: criar novas tarefas, apagar as antigas, atualizar/deletar a instância, resolver ramos de join, e registrar operações de variável — tudo com a garantia de que ou tudo é aplicado, ou nada é.

Implementação MongoDB: transações nativas

MongoKikwiEngineRepository.commitWork usa ClientSession.withTransaction do driver nativo do MongoDB — transações multi-documento ACID reais, não uma simulação em nível de aplicação:

try (ClientSession clientSession = mongoClient.startSession()) {
    clientSession.withTransaction(() -> {
        // criar/atualizar/deletar process_instances
        // inserir/remover executable_tasks e external_tasks
        // criar incidentes, resolver branchPullIntentions via updateOne com pipeline agregado
        return "Transaction committed";
    });
}

Requisito de infraestrutura: o MongoClient precisa apontar para um replica set (mesmo que de um único nó) — transações multi-documento do MongoDB não funcionam contra uma instância standalone. Isso é transparente para quem usa o kikwi-runtime-persistence-mongodb-spring-boot-starter, mas é uma pré-condição de infraestrutura que qualquer time de plataforma precisa garantir antes de colocar o Kikwiflow em produção com Mongo.

As cinco coleções

Coleção	Conteúdo
process_definitions	Versões implantadas de ProcessDefinition, indexadas por key+checksum (idempotência de deploy) e por key (última versão).
process_instances	Estado corrente de cada ProcessInstance ativa — variáveis, status, nós ativos. Instâncias COMPLETED são deletadas, não arquivadas nesta coleção (ver nota abaixo).
executable_tasks	Fila de trabalho interno: tarefas PENDING/LOCKED/AWAITING_BRANCHES, incluindo timers e joins.
external_tasks	Tarefas aguardando conclusão externa (workers, ação humana).
incidents	Registros OPEN/RESOLVED gerados pelo FailureHandler.
outbox_events	Eventos críticos (FLOW_NODE_FINISHED, GATEWAY_ANSWER_RESOLVED, PROCESS_INSTANCE_FINISHED), persistidos quando kikwiflow.outbox.events-enabled=true. Ver Eventos e Observabilidade.

:::info Instâncias concluídas são removidas de process_instances, mas não do histórico Em ContinuationService.handleContinuation, quando uma instância atinge ProcessInstanceStatus.COMPLETED, ela é marcada como instanceToDelete no UnitOfWork (se já estava persistida) — ou seja, o registro em process_instances é apagado, não apenas atualizado para status COMPLETED. Isso não é um problema para auditoria/histórico porque a coleção outbox_events (com kikwiflow.outbox.events-enabled=true) preserva o evento PROCESS_INSTANCE_FINISHED — e todo o rastro de FLOW_NODE_FINISHED/GATEWAY_ANSWER_RESOLVED da instância — indefinidamente, mesmo depois que o registro "quente" em process_instances deixa de existir. Ver Eventos e Observabilidade. :::

Resolução de joins via pipeline de agregação atômico

A remoção de um branchId concluído de um JOIN_GATEWAY pendente (pendingBranchIds) é feita com um único updateOne usando um pipeline de agregação MongoDB — remove o branchId do array e, na mesma operação atômica, recalcula status: PENDING se o array ficou vazio:

new Document("$set", new Document("pendingBranchIds",
        new Document("$filter", new Document("input", "$pendingBranchIds")
                .append("as", "b")
                .append("cond", new Document("$ne", List.of("$$b", branchId))))));

new Document("$set", new Document("status",
        new Document("$cond", new Document("if", new Document("$eq", List.of(new Document("$size", "$pendingBranchIds"), 0)))
                .append("then", "PENDING")
                .append("else", "$status"))));

Isso evita uma leitura-modificação-escrita (read-modify-write) clássica, que exigiria lock otimista com retry em caso de dois ramos concluindo "ao mesmo tempo" — o próprio MongoDB resolve a concorrência dentro da operação atômica.

Locking distribuído de tarefas: findAndLockDueTasks

Coberto em detalhe em Workers e Tarefas Externas — este é o mecanismo que permite múltiplas instâncias da aplicação (escala horizontal) competirem de forma segura pelas mesmas tarefas pendentes, sem duas instâncias processarem a mesma ExecutableTask simultaneamente.

Escrita de variáveis: delta, não snapshot completo

ProcessInstanceExecution.getVariables() retorna um Map decorado que intercepta put/putAll/remove e acumula cada operação como um VariableOperation (SET/UNSET) em vez de simplesmente mutar um mapa em memória:

public enum VariableOpType { SET, UNSET }
public record VariableOperation(ProcessVariable value, VariableOpType type) {}

Esse acumulador (variableOperations) é drenado para o UnitOfWork a cada handleContinuation, permitindo que a camada de persistência aplique apenas o delta de variáveis alteradas naquele passo — em vez de sobrescrever o documento inteiro da instância a cada transição de nó. Isso importa em dois cenários: reduz o volume de dados escrito por transição (relevante para instâncias com muitas variáveis grandes), e reduz a superfície de conflito em cenários de concorrência (duas escritas concorrentes em variáveis diferentes da mesma instância não colidem).

Índices

KikwiEngineRepository.ensureIndexes() é chamado no bootstrap da aplicação para garantir que os índices necessários (busca por key+checksum em definições, busca de tarefas devidas por dueDate, etc.) existam — combinado com auto-index-creation: true no application.yml do driver Spring Data MongoDB.