Varias réplicas

​

Cuándo la necesitas (y cuándo no)

​

Configuración

import { Bot } from '@botmesh/core'
import { BaileysProvider } from '@botmesh/baileys'
import { RedisStore } from '@botmesh/redis-store'

const store = new RedisStore({ url: process.env.REDIS_URL, prefix: 'tienda' })

const bot = new Bot({
  provider: new BaileysProvider({ session: 'tienda' }),
  store,                                     // estado compartido: chats, checkpoints, ctx.user
  coordination: {
    coordinator: store.coordinator({ leaseMs: '15s' }), // mismo cliente y prefijo que el store
    // turnWaitMs: '2m',        // máximo que un mensaje espera el turno de su chat
    // settleMs: '1m',          // máximo que se retiene un turno por un paso de conversación en curso
    // providerSession: 'auto', // 'leader' | 'shared' | 'auto'
    // providerDownMs: '1m',    // una líder con el transporte cerrado este tiempo cede el puesto
  },
})
await bot.start()

​

Qué obtienes

​

Relevo de conversaciones

const bot = new Bot({
  provider, store,
  coordination: {
    coordinator,
    handover: {
      flows: 'forward',      // guarda también los flujos sin .persistent() (con TTL)
      fallback: 'notify',    // el resto: avisa al usuario de que se interrumpió
      ttl: '10m',            // igual o mayor que conversation.timeout
    },
  },
})
bot.on('coordination:handover', ({ key, name, outcome }) => log.info({ key, name, outcome }, 'relevo'))

• flows: 'forward' escribe un checkpoint por paso también para los flujos sin .persistent(), marcado como efímero: caduca ttl después de su última escritura (por defecto conversation.timeout, si no 5 minutos; mientras el flujo espera, nunca antes del timeout de esa espera, y se renueva en cada espera) y lleva una huella del flujo, para que una réplica con el flujo cambiado (despliegue gradual) no lo retome en el paso equivocado; ese caso va a fallback con reason: 'changed'. c.state tiene que ser JSON plano: un flujo cuyo estado guarda una instancia de clase, una función o un Date no se reenvía y va a fallback. Tras un bot.stop() ordenado el checkpoint se conserva, así que la réplica que toma el relevo (o esta misma, reiniciada dentro del TTL) sigue. Actualiza todas las réplicas antes de activar forward (y desactívalo antes de volver atrás): una réplica que no reenvía igualmente retoma un checkpoint efímero y lo borra, pero una versión anterior ignora su caducidad.
• fallback se aplica a lo que no se puede serializar (la pila y los closures de una función asíncrona). La conversación deja un pequeño marcador en el store mientras espera (dos escrituras por conversación: una cuando espera por primera vez y otra para borrarlo). La réplica que toma el chat entonces:
  • 'notify': responde notifyMessage (por defecto, el texto localizado core.conversation.interrupted) y consume el mensaje;
  • 'restart': vuelve a entrar en la conversación por nombre, desde el principio y con su estado inicial (lo que se ejecutó antes de su primera espera se repite);
  • una función (info, ctx) => boolean: tu migración; devuelve true si consumió el mensaje;
  • 'drop' (por defecto): nada, el mensaje se enruta con normalidad.

• La unidad del relevo es el paso. Un paso que se repite en la réplica nueva repite sus efectos anteriores a su primera espera: los envíos solo se deduplican con durability, y el trabajo de .do() es al-menos-una-vez.
• Las respuestas a preguntas posteriores de un paso con varias preguntas (un formulario en modo texto, un .do() con varios c.ask) no se reenvían: el paso vuelve a empezar.
• Sin durability, el envío de la pregunta y su marca no son atómicos: una caída entre ambos pregunta una vez más. Las respuestas mayores que maxAnswerBytes no se incrustan (el paso vuelve a preguntar).
• El timeout de una espera vuelve a empezar en la réplica nueva, acotado por el TTL. Los vigilantes c.idle() solo se rearman con el código que se repite.
• Una conversación suspendida antes de su primera espera no deja marcador (todavía no hay nada que relevar).
• Las conversaciones asíncronas nunca se migran solas.
• Todos los stores integrados escriben los checkpoints con compare-and-set (ver arriba). Un store propio sin setCheckpointIf/clearCheckpointIf solo tiene el fencing local: una comprobación justo antes de cada escritura, como en los envíos.

​

Requisitos y límites

• Un store compartido: RedisStore, PostgresStore, MysqlStore o MongoStore (solo Redis trae coordinador; los demás guardan el estado compartido y los checkpoints con compare-and-set). Con MemoryStore o SQLiteStore cada réplica tiene sus propios datos.
• Los flujos de varios turnos deberían ser .persistent() (o usa el handover). Sin eso, un bot.conversation o un flujo no persistente vive en la memoria de la réplica que lo inició; si el siguiente mensaje llega a otra réplica, se enruta como un mensaje nuevo. Con coordinación un flujo persistente se guarda en cuanto se entra en él, así que incluso su primera pregunta sobrevive a un cambio de réplica.
• El dedupe de entrada es por proceso. Un reintento de webhook que llegue a otra réplica se procesa otra vez. Haz los handlers idempotentes, deja el transporte en una sola réplica ('leader') o activa durability con un backend compartido: su journal de entrada deduplica entre réplicas.
• Un mensaje que espera turno en una réplica que muere se pierde (como en un único proceso que se cae), salvo con durability y un backend compartido: el mensaje quedó anotado al llegar y otra réplica lo reprocesa cuando vence su préstamo.
• Las tareas programadas simples (bot.schedule/every sin id) corren en todas las réplicas. Conviértelas en tareas durables, que ejecutan cada ocurrencia una sola vez, o toma tú el liderazgo (ver Tareas únicas).
• Un paso de conversación más largo que settleMs se interrumpe (ver arriba): sube settleMs para trabajo largo en .do() o delays.
• Redis: un failover que pierda las últimas escrituras puede conceder un lease dos veces (standalone con Sentinel, o Redis Cluster). Redis Cluster necesita el layout de claves 'cluster': ver Redis Cluster.

​

Cómo elegir leaseMs

​

Eventos

bot.on('coordination:leader', ({ name, fence }) => log.info({ name, fence }, 'transporte conectado'))
bot.on('coordination:follower', ({ name, reason }) => log.warn({ name, reason }, 'en espera'))
bot.on('coordination:lease-lost', ({ key, reason }) => log.warn({ key, reason }, 'lease perdido'))
bot.on('coordination:handover', ({ key, name, outcome }) => log.info({ key, name, outcome }, 'conversación relevada'))
bot.on('error', ({ error }) => {
  // CoordinationUnavailableError: un mensaje no consiguió su turno (timeout o Redis caído)
})

​

Tareas únicas y locks con fencing

const coordinator = store.coordinator()
bot.every('5m', async () => {
  const lead = await coordinator.tryLead('job:informe')   // null en las demás réplicas
  if (!lead) return
  try { await enviarInforme() } finally { await lead.release() }
})

​

Docker Compose con dos réplicas

services:
  redis:
    image: redis:7-alpine
  bot:
    build: .
    environment:
      REDIS_URL: redis://redis:6379
    volumes:
      - ./sessions:/app/sessions   # Baileys: la líder necesita los archivos de sesión
    deploy:
      replicas: 2
    depends_on: [redis]

​

Tests

import { Bot, MemoryStore } from '@botmesh/core'
import { MemoryCoordinationHub, MemoryCoordinator } from '@botmesh/core/testing'

const hub = new MemoryCoordinationHub()
const store = new MemoryStore()
const a = new Bot({ provider: pa, store, coordination: { coordinator: new MemoryCoordinator({ hub, replicaId: 'a' }) } })
const b = new Bot({ provider: pb, store, coordination: { coordinator: new MemoryCoordinator({ hub, replicaId: 'b' }) } })