@tuain/bpm-engine

Motor de gestión de procesos de negocio (BPM) de la plataforma Tuain.

SOFTWARE PROPIETARIO, DE DISTRIBUCIÓN ABIERTA. Copyright (c) 2022-2026 Imix Latam / TUAIN. Todos los derechos reservados. Desde la 2.4.0 el paquete se publica con acceso público y se instala y ejecuta sin token de ningún tipo, pero eso es una decisión de distribución, no un cambio de licencia: no es software libre ni de código abierto. Su uso sigue sujeto al archivo LICENSE — licencia de uso no exclusiva, no transferible y revocable, sin derecho a modificar, redistribuir, sublicenciar, descompilar ni aplicar ingeniería inversa.

Qué es

Motor de ejecución de procesos definidos como tareas, gateways y eventos, con persistencia en MongoDB/DocumentDB. Se compone de dos piezas:

• ProcessManager — administra las definiciones (catálogo de procesos, tareas, gateways, eventos, variables y roles).
• ProcessController — ejecuta instancias de un proceso. Se extiende en el servicio consumidor, que registra el comportamiento del flujo en customizeBehaviour().

El paquete se distribuye como un único bundle CommonJS minificado (dist/index.js) con sus definiciones de tipos (dist/index.d.ts). El código fuente no se publica.

Instalación

npm install @tuain/bpm-engine

Desde la 2.4.0 no hace falta nada más: el paquete se publica con acceso público, así que no se necesita .npmrc ni token de lectura. Si tu repo trae un .npmrc con //registry.npmjs.org/:_authToken puesto solo para este paquete, ya puedes quitarlo.

Las versiones 2.3.x y anteriores sí requerían token de lectura del scope @tuain.

Licencia de ejecución

Desde la 2.4.0 el motor no exige licencia: la verificación viene inerte y el arranque no requiere TUAIN_BPM_LICENSE. Si la variable está presente se ignora, así que un despliegue que la traiga de la 2.3.x no falla — puedes retirarla cuando quieras.

El motor sabe verificar una licencia Ed25519 firmada por el titular, presentada en TUAIN_BPM_LICENSE o como opción del constructor:

new ProcessManager(ecosystem, getDb, getDb, logger, errMgr, { license: miLicencia });

La verificación es offline: no hay llamadas a servicios del titular. manager.license expone el veredicto — en esta versión, siempre notConfigured. El titular puede reactivarlo en una versión posterior; si eso ocurre se anunciará como cambio de comportamiento en el CHANGELOG, no de forma silenciosa.

Uso

const { ProcessManager, ProcessController } = require('@tuain/bpm-engine');

const manager = new ProcessManager(ecosystem, getDb, getDb, logger, errMgr);

class MiProceso extends ProcessController {
  customizeBehaviour() {
    this.onTaskStart('miTarea', this.iniciarMiTarea.bind(this));
  }
  async iniciarMiTarea({ procId, id }) {
    await this.setVars(procId, { algo: true });
    this.completeTask(id);
  }
}

const proceso = new MiProceso('miProceso', manager, definicion);

Toda operación retorna la tupla [error, valor].

Arranque

El constructor no puede esperar la carga de la definición, así que la inicialización se expone en ready y conviene esperarla antes de empezar a atender tráfico:

const [initErr] = await proceso.ready; // nunca rechaza: informa por la tupla

Los puntos de entrada (newInstance, completeTask, triggerEvent) la esperan por su cuenta, de modo que una petición temprana espera en lugar de encontrarse sin definición ni callbacks registrados. Si la inicialización falló, esas operaciones retornan un error explícito de controlador no inicializado. proceso.initialized expone el estado.

Operación con múltiples instancias (desde 2.2.0)

El motor no tiene bucle de ejecución propio: la cascada de nodos avanza por recursión sobre el event loop de la instancia que atendió la entrada (newInstance, completeTask o el disparo de un temporizador). Varias instancias del mismo servicio pueden convivir sobre la misma base de datos, con dos condiciones:

Temporizadores. Cada instancia agenda todos los timers programados, así que la exclusión se resuelve al disparar: triggerTimerEvent reclama el timer con un compare-and-swap de scheduled a executed sobre la colección timers y solo la ganadora ejecuta los callbacks y avanza el flujo. No requiere índices nuevos. La colección TimerLocks quedó sin uso (su índice nunca fue unique, por lo que no excluía nada) y puede eliminarse.

Los timers reclamados registran claimedBy (instancia), execution (fecha), attempts (reclamos consumidos), leaseExpiresAt (vigencia del reclamo) y flowAdvanced.

Recuperación de temporizadores estancados (desde 2.5.0). El reclamo es exclusivo pero ya no es definitivo. Si la instancia que lo gana no alcanza a completar el avance —la mataron a mitad del despliegue, un callback del consumidor reventó o quedó colgado, la base de datos falló entre una escritura y la siguiente— el temporizador no queda detenido para siempre:

• Cuando el motor ve la falla (excepción o error de goToNode), devuelve el temporizador a scheduled con la fecha de despertar corrida por timerRetryBackoff (1 minuto) y lo reagenda de inmediato. El documento queda con lastFailure.
• Cuando no la ve —la instancia dejó de existir—, la barrida recoverStalledTimers() lo devuelve a scheduled al vencer la concesión (timerLeaseTimeout, 10 minutos). Corre en cada ciclo de mantenimiento, junto al refresco, y recupera hasta timerRecoveryBatch (50) por pasada.
• Los callbacks del evento no se repiten en el reintento: el temporizador queda con callbacksDone: true y el segundo reclamo solo reintenta el avance del flujo.
• Tras timerMaxAttempts (5) reclamos fallidos el temporizador queda en status: 'failed' con el motivo en failure — una falla determinista no se reintenta para siempre. Es el estado a vigilar: significa que ese proceso necesita intervención manual.

Temporizadores abandonados (desde 2.9.0). Un temporizador cuyo wakeupDate pasó sin que ninguna instancia estuviera viva para dispararlo no se dispara después: al agendar se descarta si lleva más de timerAbandonAfterHours (24) vencido, para que levantar el servicio tras un fin de semana no lance una cascada de cientos de llamadas contra los servicios externos. Hasta la 2.8.0 ese descarte era una evaluación en memoria: el documento seguía scheduled y el refresco lo volvía a leer y a reportar en warn cada cinco minutos, indefinidamente.

Ahora abandonExpiredTimers() lo marca abandoned (con abandonedAt y abandonedBy) y el lote se reporta en una sola línea con el rango de fechas y el desglose por evento. Va dentro de timerCleanup(), que corre al inicio de cada ciclo periódico y se desprende en el arranque: la limpieza no retrasa ready. No se borran: cada abandoned señala una instancia de proceso detenida en ese nodo, que es la evidencia con la que se diagnostica. La purga existe (timerPurgeAbandonedAfterDays) pero viene desactivada.

Temporizadores cancelables (desde 2.10.0). Un temporizador pendiente se puede dar de baja, y la instancia que termina se lleva los suyos. Hasta la 2.9.0 no había forma de cancelarlo: un plazo seguía corriendo aunque el proceso ya no lo necesitara, y los temporizadores de una instancia terminada despertaban igual, ejecutaban los callbacks del evento sobre un proceso cerrado y quedaban en failed. Ahora hay tres formas de cancelar, todas dejan status: 'cancelled' con cancellation: { at, by, reason }:

• Desde la definición: un nodo (tarea, gateway o evento) con cancelTimers: ['nombreEvento', …] cancela esos temporizadores al llegar el flujo a él, antes de crear el nodo.
• Desde el consumidor: controller.cancelTimers(procId, names?, reason?), para lo que no pasa por el flujo (por ejemplo, otro módulo que decide que un plazo deja de correr).
• Al terminar la instancia: el evento end cancela todos sus temporizadores pendientes. Y si un temporizador despierta sobre una instancia ya terminada —uno creado por una versión anterior, o una carrera con el cierre—, se cancela sin ejecutar callbacks ni avanzar.

La cancelación es atómica frente al disparo: los dos parten de scheduled, así que de una cancelación y un disparo simultáneos gana exactamente uno. Un plazo que debe correr en paralelo al flujo se modela con un gateway paralelo al inicio: una rama es el flujo normal y la otra el temporizador, cuyo next es un evento end (vencido); los nodos que desactivan el plazo llevan cancelTimers.

El motor crea por su cuenta el índice stalledTimersRecovery (status, flowAdvanced, leaseExpiresAt) sobre timers la primera vez que barre. Los temporizadores executed de versiones anteriores, que no escribían flowAdvanced, quedan fuera de la barrida y no se re-disparan.

Un controlador por proceso (desde 2.6.0). El refresco y la barrida de temporizadores están acotados por ecosystem y processName. Antes no lo estaban: como un servicio instancia un controlador por proceso sobre las mismas colecciones, cada controlador agendaba y reclamaba los temporizadores de todos los demás, y el que ganaba el reclamo no tenía ese evento en su definición — el temporizador quedaba consumido y el controlador dueño no lo disparaba nunca. Si tu servicio hospeda más de un proceso BPM, esto afectaba a todos.

Recuperación del avance de tareas y gateways (desde 2.6.0). Cerrar un nodo y crear el siguiente son operaciones separadas y sin transacción. El cierre es atómico y excluyente, pero de ahí al nodo siguiente había un tramo sin red: la tarea quedaba complete sin sucesor, o el join complete sin destino disparado, y nada volvía a mirarlos.

Ahora el cierre escribe, en su misma operación atómica, un marcador de avance pendiente (flowAdvanced: false, advancedBy, advanceLease, advanceAttempts), y recoverStalledNodes() retoma lo que quede vencido — en cada ciclo de mantenimiento y también al arrancar el controlador, porque un reinicio es justo lo que deja nodos a medias:

• No se rehace el cierre. Una tarea completada no se reabre ni repite sus callbacks: la recuperación solo crea el nodo siguiente. Es la diferencia con los temporizadores, que sí se re-encolan completos.
• Los gateways registran el progreso destino por destino (destinations.<nombre>.advanced), así que un avance a medias se retoma sin repetir los destinos ya creados.
• Tras advanceMaxAttempts (5) intentos el nodo queda con advanceFailed: true y el motivo en advanceFailure. Es el otro estado a vigilar, junto con los timers en failed. status no se toca, para no alterar las consultas de historial del consumidor.
• El gateway estándar ahora se cierra después de disparar sus destinos y en la misma escritura que su marcador; antes el cierre iba en paralelo con ellos y podía quedar complete con destinos sin crear.

Índice stalledNodeAdvanceRecovery (ecosystem, procName, flowAdvanced, advanceLease) sobre tasks y gateways, creado por el motor. Los nodos cerrados por versiones anteriores no tienen marcador y quedan fuera de la barrida.

Consultas de vigilancia:

db.TasksInstances.countDocuments({ flowAdvanced: false, advanceFailed: { $exists: false } }); // pendientes
db.TasksInstances.countDocuments({ advanceFailed: true }); // agotados: requieren intervención
db.GatewaysInstances.countDocuments({ advanceFailed: true });
db.timers.countDocuments({ status: 'failed' });

Procesos parados antes de la 2.5.0. No traen marcador, así que el motor no puede distinguirlos de los que terminaron bien y no los recupera solo. Se ubican con el script de diagnóstico del repositorio del motor (no viaja en el paquete), que reconoce a una instancia detenida por no tener nada que la mueva —ninguna tarea activa, ningún temporizador programado, ningún join esperando ramas— y clasifica el caso por su último nodo:

# Reporte, sobre el driver de Node: no requiere mongosh y es de solo lectura sin excepción
node scripts/report-stalled-processes.mjs --uri "<uri>" --ecosystem <eco>

# Con mongosh, si se prefiere el shell — y el único camino que puede reparar
mongosh "<uri>" scripts/locate-stalled-processes.mongo.js                                  # solo lectura
mongosh "<uri>" --eval 'bpmStalledConfig = { APPLY: true }' -f scripts/locate-stalled-processes.mongo.js

El lanzador de Node acepta --db, --process, --stale-minutes, --limit y --tls-ca-file (para DocumentDB), o la URI por MONGO_URI. Carga los documentos y corre el mismo diagnóstico contra un doble síncrono, así que el reporte sale de la lógica que tiene regresión y no de una copia.

Con APPLY escribe el marcador que lee la barrida del motor —no reimplementa el avance— y solo en los casos donde no puede duplicar un avance que ya ocurrió; el resto los reporta para revisión. Es idempotente y tiene tope por corrida. Requiere el servicio en 2.6.0 o superior. La configuración (ecosistema, proceso, umbral de inactividad, topes) está al inicio del archivo.

Instancias de tarea. La instancia activa de una tarea se crea por upsert con clave de exclusión, respaldada por un índice único sparse sobre activeKey (uniqueActiveTaskInstance, creado por el propio motor la primera vez que instancia una tarea). Desde la 2.8.0 la clave es por nodo + llegada —incluye el origin que la produjo—, igual que la de los gateways que no son join: un avance re-disparado converge sobre la instancia que ya existe, mientras que el mismo nodo alcanzado de nuevo en una vuelta posterior de un bucle crea su propia instancia. La clave se libera al completar, cerrar o cancelar la tarea. Las instancias creadas por versiones anteriores no tienen la clave y se siguen reutilizando con normalidad.

Con la clave por nodo —el comportamiento hasta la 2.7.0— la llegada de la vuelta N+1 de un bucle se fundía con la instancia todavía activa de la vuelta N: no se creaba instancia, no corría el callback de inicio y el token desaparecía sin excepción, sin log de error y sin quedar como candidato de ninguna barrida. Bastaba una instancia que quedara activa sin completarse para envenenar el nodo: de ahí en adelante toda vuelta convergía sobre el cadáver y el proceso quedaba detenido.

convergeArrivals (desde 2.8.0). Una tarea que declara convergeArrivals: true recupera la clave por nodo, y con ella la convergencia entre llegadas distintas: es el modo para los nodos de convergencia legítima, donde varias ramas concurrentes de un fork llegan al mismo nodo y deben fundirse en una sola instancia con un solo callback de inicio. Hay que declararlo nodo por nodo, y solo donde las llegadas puedan ser simultáneas de verdad: las alternativas de un gateway exclusivo no lo necesitan.

tasks: {
  notificacionFalloAfiliacion: { type: 'sync', convergeArrivals: true, next: { … } },
}

Temporizadores de un evento (desde 2.8.0). registerTimerEvent recibe la llegada cuando lo invoca el avance del flujo, y registra el temporizador por upsert sobre la misma clave de nodo + llegada, con su índice único sparse (uniqueTimerArrival). Así un avance re-disparado no programa un segundo disparo del mismo evento —dos temporizadores son dos documentos, y el reclamo por _id no los cruza, de modo que el flujo avanzaría dos veces desde el mismo punto—, mientras que la vuelta siguiente del bucle sí programa el suyo. Registrarlo sin llegada (el consumidor llamando por su cuenta) conserva la inserción directa de siempre.

El callback de inicio corre una sola vez por instancia (desde 2.7.0). onTaskStart corría en cada llegada, también cuando la rama convergía sobre una instancia ya existente: en un join o un fork se ejecutaba una vez por rama, duplicando los efectos externos que hubiera dentro. Ahora solo corre cuando la instancia se creó de verdad. Si tu callback dependía de correr por cada llegada, hay que moverlo a onTaskComplete o a la lógica del gateway.

Instancias de gateway (desde 2.7.0). Los gateways que no son join —estándar y fork— también se crean con exclusión mutua, con la clave formada por el nodo y la llegada que lo produjo (origin). Así un avance re-disparado —lo que hace la recuperación al retomar el nodo anterior— converge sobre la instancia existente en lugar de insertar una segunda y duplicar la rama completa aguas abajo, mientras que una llegada distinta, o el mismo nodo alcanzado de nuevo en un ciclo, sigue creando su propia instancia. La clave se libera al cerrar el gateway. Un efecto colateral visible: los gateways fork ahora quedan en complete al terminar de disparar sus destinos; antes se quedaban active para siempre y no aparecían en getProcessGatewaysHistory.

Finalización de tareas. completeTask reclama la tarea (active → complete) antes de ejecutar los callbacks y avanzar el flujo, en una sola operación atómica. Así un doble envío desde la UI, un reintento del cliente ante un timeout o la misma petición atendida por dos instancias no duplican la cascada: quien pierde el reclamo retorna [null, { alreadyCompleted: true }], sin efectos. La tarea queda con completedBy para saber qué instancia la completó. No requiere índices nuevos.

Cierre ordenado. Ante la señal de terminación hay que invocar await controller.shutdown() por cada controlador: cancela el refresco y los temporizadores agendados en memoria, y espera a que terminen las cascadas ya desprendidas del flujo (plazo shutdownDrainTimeout, 15 s por defecto). Sin esa llamada, un despliegue normal mata cascadas a medio camino y deja procesos parqueados en el nodo donde iban. El plazo debe quedar por debajo del periodo de gracia del orquestador (terminationGracePeriodSeconds en Kubernetes).

controller.pendingWork() expone cuántas operaciones desprendidas hay en vuelo; sirve como métrica de drenaje y para readiness probes.

Desarrollo

npm test                # mocha
npx eslint lib config test
npx prettier --check lib config test
npm run copyright       # refresca los encabezados de copyright en los fuentes
npm run build           # genera dist/index.js + dist/index.d.ts

npm publish ejecuta build automáticamente (prepublishOnly) y publica con acceso público.

Contacto

Imix Latam / TUAIN — titular de los derechos. Para autorizaciones de uso, contactar al titular.