Running N replicas

​

When you need it (and when you don’t)

​

Setup

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: 'shop' })

const bot = new Bot({
  provider: new BaileysProvider({ session: 'shop' }),
  store,                                     // shared state: chats, checkpoints, ctx.user
  coordination: {
    coordinator: store.coordinator({ leaseMs: '15s' }), // same client and prefix as the store
    // turnWaitMs: '2m',        // longest a message waits for its chat's turn
    // settleMs: '1m',          // longest a turn stays held for a running conversation step
    // providerSession: 'auto', // 'leader' | 'shared' | 'auto'
    // providerDownMs: '1m',    // a leader whose transport stays closed this long steps down
  },
})
await bot.start()

​

What you get

​

Conversation hand-over

const bot = new Bot({
  provider, store,
  coordination: {
    coordinator,
    handover: {
      flows: 'forward',      // checkpoint flows without .persistent() too (TTL-bound)
      fallback: 'notify',    // the rest: tell the user it was interrupted
      ttl: '10m',            // keep at or above conversation.timeout
    },
  },
})
bot.on('coordination:handover', ({ key, name, outcome }) => log.info({ key, name, outcome }, 'hand-over'))

• flows: 'forward' writes a checkpoint per step for flows without .persistent() too, marked ephemeral: it expires ttl after its last write (default conversation.timeout, else 5 minutes; while the flow waits, never before that wait’s own timeout, refreshed at every wait) and carries a fingerprint of the flow, so a replica running a changed flow (rolling deploy) does not resume it at the wrong step; that case goes to fallback with reason: 'changed'. c.state must be plain JSON: a flow whose state holds a class instance, a function or a Date is not forwarded and goes to fallback. After a graceful bot.stop() the checkpoint is kept, so the replica that takes over (or this one, restarted within the TTL) continues. Upgrade every replica before turning forward on (and turn it off before rolling back): a replica that does not forward still resumes an ephemeral checkpoint and clears it, but an older version ignores its expiry.
• fallback applies to what cannot be serialized (an async function’s stack and closures). The conversation keeps a small stub in the store while it waits (two writes per conversation: one when it first waits, one to clear it). The replica that takes the chat then:
  • 'notify': replies notifyMessage (default: the localized core.conversation.interrupted text) and consumes the message;
  • 'restart': enters the conversation again by name, from the beginning, with its initial state (what ran before its first wait runs again);
  • a function (info, ctx) => boolean: your migration; return true when it consumed the message;
  • 'drop' (default): nothing, the message routes normally.

• The unit of hand-over is the step. A step run again on the new replica runs its side effects before its first wait again: sends are deduplicated only with durability, and .do() work is at-least-once.
• Answers to later questions of a multi-question step (a text-mode form, a .do() with several c.ask) are not forwarded: the step restarts.
• Without durability, the prompt send and its marker are not atomic: a crash in between asks once more. Answers larger than maxAnswerBytes are not embedded (the step asks again).
• A wait’s timeout restarts on the new replica, bounded by the TTL. c.idle() watchdogs are armed again only by the code that runs again.
• A conversation suspended before its first wait leaves no stub (nothing to hand over yet).
• Async conversations are never migrated automatically.
• Every built-in store writes checkpoints with a compare-and-set (see above). A custom store without setCheckpointIf/clearCheckpointIf only gets the local fence: a check right before each write, like sends.

​

Requirements and limits

• A shared store: RedisStore, PostgresStore, MysqlStore or MongoStore (only Redis has a coordinator; the others hold the shared state and the checkpoints with compare-and-set). With MemoryStore or SQLiteStore each replica has its own data.
• Multi-turn flows should be .persistent() (or use handover). Without it, a plain bot.conversation or non-persistent flow lives in the memory of the replica that started it; if the next message reaches another replica it is routed as a fresh message. With coordination a persistent flow is checkpointed as soon as it is entered, so even its first question survives a hand-over.
• Inbound dedupe is per process. A webhook retry that reaches another replica is processed again. Make handlers idempotent, keep the transport on one replica ('leader'), or turn on durability with a shared backend: its inbox journal deduplicates across the replicas.
• A message waiting for its turn on a replica that dies is lost (as in a single process that crashes), unless durability is on with a shared backend: the message was journaled on arrival and another replica replays it once its lease expires.
• Plain scheduled jobs (bot.schedule/every without an id) run on every replica. Make them durable tasks, which run each occurrence once, or take the leadership yourself (see Singleton jobs).
• A conversation step longer than settleMs is cut short (see above): raise settleMs for long .do() work or delays.
• Redis: a failover that loses the last writes can grant a lease twice (standalone with Sentinel, or Redis Cluster). Redis Cluster needs the 'cluster' key layout: see Redis Cluster.

​

Choosing leaseMs

​

Events

bot.on('coordination:leader', ({ name, fence }) => log.info({ name, fence }, 'connected the transport'))
bot.on('coordination:follower', ({ name, reason }) => log.warn({ name, reason }, 'standing by'))
bot.on('coordination:lease-lost', ({ key, reason }) => log.warn({ key, reason }, 'lease lost'))
bot.on('coordination:handover', ({ key, name, outcome }) => log.info({ key, name, outcome }, 'conversation handed over'))
bot.on('error', ({ error }) => {
  // CoordinationUnavailableError: a message could not get its turn (timeout or Redis down)
})

​

Singleton jobs and fenced locks

const coordinator = store.coordinator()
bot.every('5m', async () => {
  const lead = await coordinator.tryLead('job:report')   // null on the other replicas
  if (!lead) return
  try { await sendReport() } finally { await lead.release() }
})

​

Docker Compose with two replicas

services:
  redis:
    image: redis:7-alpine
  bot:
    build: .
    environment:
      REDIS_URL: redis://redis:6379
    volumes:
      - ./sessions:/app/sessions   # Baileys: the leader needs the session files
    deploy:
      replicas: 2
    depends_on: [redis]

​

Testing

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' }) } })