Procesamiento durable de mensajes

import { Bot } from '@botmesh/core'
import { SQLiteStore } from '@botmesh/sqlite-store'

const store = new SQLiteStore({ file: './data/bot.db' })

const bot = new Bot({
  provider,
  store,
  durability: {
    backend: store.durability(),       // journals en el mismo archivo SQLite
    instanceId: 'bot-1',               // estable entre reinicios de este despliegue
    outbox: { retries: 5, backoff: { baseMs: 1000, capMs: 60_000 } },
  },
})

• El inbox. Cada mensaje o interacción admitido se anota antes de que corra ningún handler, y se marca como hecho solo cuando sus efectos se asentaron. Tras una caída, el siguiente arranque reprocesa lo que quedó a medias.
• El outbox. Cada envío se anota con una clave de idempotencia antes de llamar al provider, y justo después se guarda el id de mensaje que devolvió el provider. Un envío cuya clave ya está guardada se responde desde el outbox en vez de salir otra vez. Así un reproceso no manda dos veces la misma respuesta. Los envíos que fallan se reintentan con backoff y van a la cola de muertos (DLQ) cuando se les acaban los reintentos.

​

Backends

​

Un journal de @botmesh/events

import { FileEventJournal } from '@botmesh/events'
import { eventsDurability } from '@botmesh/events/durability'

const journal = await FileEventJournal.open('./data/events.json', { lock: true })
const bot = new Bot({ provider, durability: { backend: eventsDurability({ journal }) } })
// después de bot.stop(): await journal.close()

• Los registros de la inbox son EventRecords de ese journal, en el canal 'botwa:durable', con eventType 'botwa.inbound.message' o 'botwa.inbound.interaction'. El list() y el replay() del journal los ven, y el barrido del bot recoge un replay. Un EventWorker nunca los toma: un claim sin canal solo toma registros sin canal.
• El outbox es el del core, guardado como sidecar del journal: el mismo archivo y el mismo snapshot atómico que la inbox.
• Un FileEventJournal debe abrirse con { lock: true }. El archivo de lock con el pid impide que un segundo proceso lo abra, y un segundo open() del mismo archivo en el mismo proceso se rechaza hasta que se cierra el primero.
• Coste: un FileEventJournal reescribe el archivo entero en cada escritura, y aquí ese archivo guarda los eventos de la app, los registros de entrada del bot (durante la retención, 7 días por defecto) y el outbox. Cada mensaje entrante y cada envío cuestan unas pocas reescrituras (las escrituras concurrentes comparten una, y los heartbeats de lease solo se escriben con el siguiente cambio). Para algo más que poco tráfico, da al bot un archivo de journal propio, acorta la retención o pasa un outbox respaldado por una base de datos; o usa FileDurability o un backend de base de datos.
• Para varias réplicas hace falta un LeasedEventJournal compartido, con operaciones atómicas y con fencing. SharedFileEventJournal es uno: varios procesos abren el mismo archivo y cada escritura corre bajo un mutex entre procesos (<archivo>.mutex) sobre el archivo actual, así que los claims son exclusivos y los fencing tokens valen entre procesos. El outbox vive en el mismo archivo y también se comparte:
  import { SharedFileEventJournal } from '@botmesh/events'
  
  const journal = await SharedFileEventJournal.open('./data/events.json')
  const bot = new Bot({ provider, durability: { backend: eventsDurability({ journal }) } })
  
  Está pensado para pocos procesos en un mismo host (o un sistema de archivos con rename atómico y creación exclusiva; no NFS). Por defecto cada escritura relee y reescribe el archivo entero; con appendLog: true (o { compactEvery, compactBytes, fsync }) cada escritura añade una línea a <archivo>.log con los registros/estado lateral que cambiaron, y el snapshot se reescribe (con logSeq) y el log se trunca cada 1000 entradas o 4 MiB. Un fallo entre el snapshot y el truncado es inofensivo (el replay salta las entradas ya incluidas) y una última línea a medias se recorta. Un mutex que dejó un proceso muerto se rompe al momento; un titular vivo refresca su mtime (heartbeat, heartbeatMs por defecto un tercio de staleLockMs), así que solo se da por abandonado un titular atascado tras staleLockMs (30 s por defecto). Nunca se mata nada (déjalo muy por encima de la escritura más lenta: si tres procesos compiten por romper un mutex ya atascado, dos pueden tenerlo un instante y perderse una escritura; usa un journal de base de datos si eso importa). close() deja terminar las operaciones ya llamadas y rechaza las posteriores. El formato del archivo es el de FileEventJournal, pero no abras un archivo con los dos a la vez (el journal compartido rechaza un archivo que tiene un FileEventJournal vivo con lock, también si ese journal lo abre después: cada operación compartida se rechaza en lugar de perder registros). La regla del mismo chat de Varias réplicas sigue valiendo. Para mucho tráfico, usa otro LeasedEventJournal con un outbox compartido (eventsDurability({ journal, outbox: kv }), con el DurableKv de un backend de Postgres o Redis), o esos backends directamente.

​

Configuración

​

Qué se garantiza

​

Claves de idempotencia

1. options.idempotencyKey cuando la pasas: bot.send(chatId, texto, { idempotencyKey: 'factura:42' }).
2. Si no, mientras se atiende un mensaje entrante: <id del registro entrante>:<n>, el n-ésimo envío hecho para ese mensaje. Un paso de conversación usa el mensaje que lo alimentó, así que las respuestas de un flow conservan sus claves cuando el flow se reanuda.
3. Si no, una clave aleatoria. El envío igual se reintenta y sobrevive a una caída, pero una segunda llamada no se reconoce como el mismo envío.

​

Reintentos, envíos diferidos y la cola de muertos

const { key } = await bot.enqueue(chatId, 'Tu cita es mañana a las 9', {
  idempotencyKey: `recordatorio:${cita.id}`,   // un segundo enqueue de esta clave no hace nada
})

​

Operación

bot.on('durability:dead', ({ kind, id, error }) => alertar(kind, id, error))
bot.on('durability:ambiguous', ({ key, chatId }) => auditar(key, chatId))

const d = bot.durability!
await d.inbox.list({ state: 'dead' })          // la DLQ de entrada
await d.outbox.list({ state: 'dead' })         // la DLQ de salida
await d.replayInbound(id)                      // procesarlo otra vez
await d.retrySend(key)                         // enviarlo otra vez (dead, failed o unknown)
await d.outbox.discard(key)                    // olvidarlo
d.metrics()  // { inboundJournaled, inboundDuplicates, inboundReplayed, inboundDead,
             //   outboxJournaled, outboxDeduped, outboxRetried, outboxDeferred,
             //   outboxDead, outboxAmbiguous, outboxBypassed, staleLeases, outboxDiverged }

• Leases y fencing. Un registro en proceso se presta a instanceId durante inbox.leaseMs (60 s por defecto), y un envío en curso durante outbox.leaseMs (60 s por defecto; como mínimo 20 ms, o el bot rechaza la configuración). Un latido extiende ambos mientras corren el handler o la llamada al provider, así que una subida lenta nunca se toma por un envío abandonado. Si un préstamo vence (el proceso murió o se congeló), otra réplica, o esta misma tras reiniciar, toma el registro con un nuevo token de fencing. Entonces se rechaza la confirmación del dueño anterior y se cuenta en staleLeases.
• instanceId. Mantenlo estable entre reinicios de un despliegue, para que un proceso reiniciado recupere enseguida sus registros sin terminar en vez de esperar a que venzan sus préstamos. Su primer barrido tras el reinicio solo recupera los préstamos tomados antes de que se recuperara (los de la encarnación muerta, incluida una escritura que llegó tarde), nunca un envío que está haciendo él mismo. Hazlo único por réplica que corra a la vez (el nombre del pod de un StatefulSet, por ejemplo): dos procesos vivos con el mismo id se quitarían los préstamos. En un backend de un solo proceso (memoria, archivo) el valor por defecto es <hostname>/<botId>/<connectionId>. En un backend compartido (Postgres, Redis) se le añade un sufijo por arranque, para que dos procesos del mismo host nunca lo compartan; entonces un reinicio espera a que venzan los préstamos anteriores. Fíjalo explícitamente para evitar esa espera.
• Retención. Los registros done y sent se purgan tras retention (por defecto 7d). Mantenla más larga que la ventana de reentrega más larga de tus canales, porque una reentrega solo se reconoce mientras su registro existe.
• Cierre ante fallos. Si el journal no puede anotar un mensaje entrante, el mensaje no se despacha. Se emite un 'error' y el dedupe en memoria olvida el mensaje, así que una reentrega se procesa. Con failOpen: true se despacha sin journal.
• Tamaño del contenido. El contenido saliente se anota como JSON, con los Buffers en base64. Un contenido de más de outbox.maxPayloadBytes (1 MiB por defecto), o uno que no se puede codificar (un stream), se salta el outbox con un aviso. Para archivos grandes, usa media por URL o por ruta.
• Media en un reproceso. Ver Media en un reproceso.
• Ediciones, borradores e indicadores de escritura no se anotan. Las ediciones son idempotentes, y los borradores y los indicadores son pasajeros.

​

Media en un reproceso

• 'not-durable': el provider no serializó la referencia (por ejemplo, visualización única).
• 'provider-mismatch': otro provider escribió el registro.
• 'rehydrate-failed': el valor guardado no es válido, es de una versión desconocida o fue manipulado.

bot.when(IMAGE, async (ctx) => {
  try {
    const media = await loadMediaOf(ctx, { throwOnError: true })
    // …
  } catch (error) {
    if (error instanceof MediaUnavailableError) return ctx.reply('Envíame la imagen de nuevo, por favor.')
    throw error
  }
})

​

Varias réplicas

• Envía cada chat a una sola réplica a la vez (enrutado fijo por id de chat en el balanceador, o reparte los chats entre réplicas). Si dos réplicas reciben mensajes del mismo chat, la respuesta a un ask que una réplica tiene pendiente llega a la otra como un mensaje nuevo.
• La conmutación va por préstamos. Cuando una réplica muere o se congela, otra toma sus registros sin terminar en cuanto vencen sus préstamos (inbox.leaseMs) y los reprocesa. Una réplica viva conserva sus préstamos gracias al latido, tarde lo que tarde su handler.
• Da a cada réplica su propio instanceId explícito.
• Usa un backend compartido: RedisDurability o PostgresDurability. SQLiteDurability y SharedFileEventJournal solo se comparten entre procesos de un mismo host.

​

Probar la recuperación ante caídas

import { crashable, CrashableProvider, MockProvider } from '@botmesh/core/testing'

const shared = new MemoryDurability()
const a = crashable(shared)
a.crashBefore('outbox.markSent')    // se llamó al provider, el resultado nunca se anota
const botA = new Bot({ provider: new CrashableProvider(a), store, durability: { backend: a, instanceId: 'i1' } })
// … mueve botA hasta que a.crashed …
const botB = new Bot({ provider: new MockProvider(), store, durability: { backend: shared, instanceId: 'i1' } })