Distributed Counters en NATS JetStream: el enfoque CRDT que simplifica el conteo global

Introducción

Contar eventos en un sistema distribuido parece trivial: sumar uno por cada visita a una página, restar uno por cada reembolso, mostrar el total. Pero cuando escalás a decenas de regiones con miles de incrementos por segundo, el problema se vuelve complejo. Un contador relacional con bloqueo centralizado (como Redis o una base de datos) garantiza consistencia, pero cada incremento espera coordinación y un corte de red paraliza las escrituras. Netflix resolvió esto con una abstracción de Distributed Counter capaz de manejar 75.000 incrementos por segundo con latencia en milisegundos, pero requirió desarrollar un servicio propio.

La solución clásica implica replicar contadores por región (Redis, Kafka), implementar lógica de merge y tolerar latencia en el camino crítico. Sin embargo, hay un patrón repetitivo: escribir localmente, tolerar divergencia y fusionar cuando la red se recupera. Lo que los equipos necesitan es un incremento que sea orden-independiente por construcción, seguro de reintentar y convergente sin código personalizado de reconciliación. NATS Server 2.12 entrega exactamente ese primitivo.

Qué ocurrió

Con la versión 2.12 de NATS Server, JetStream introdujo el flag AllowMsgCounter para streams. Al habilitarlo, cada subject dentro del stream se convierte en un contador de enteros firmados que se incrementa mediante un header HTTP especial. No hay locks, ni consenso, ni orden requerido entre clusters, super clusters o fuentes. La clave técnica es que el servidor gestiona el contador como una estructura de datos CRDT (Conflict-Free Replicated Data Type), donde múltiples servidores pueden actualizar copias independientes y converger al mismo resultado al fusionarse.

Bajo el capó, NATS implementa un CRDT basado en la propiedad conmutativa de la suma. Cada mensaje debe incluir el header Nats-Incr con un delta entero firmado (ej: +1, -5, +1000000). El servidor aplica el delta al valor actual del subject (almacenado en el último mensaje), actualiza el cuerpo del mensaje con el nuevo total en formato JSON ({"val":"<nuevo_total>"}) y devuelve en el PubAck el valor post-incremento en el campo val. Esto elimina la necesidad de una lectura adicional para conocer el estado del contador.

La implementación usa aritmética de enteros de precisión arbitraria (big.Int) para evitar desbordamientos, y serializa los valores como cadenas JSON en lugar de números, evitando problemas de precisión en clientes JavaScript (limitados a enteros seguros de 53 bits). Además, el flag AllowMsgCounter no puede deshabilitarse una vez activado, y su coexistencia con otras configuraciones está restringida: no permite sujetos comodín (*), solo sujetos literales o rangos explícitos.

Impacto para DevOps / Infraestructura / Cloud

Para equipos de infraestructura

La adopción de contadores distribuidos en NATS elimina la necesidad de:

• Servicios personalizados de agregación: ya no se requieren pipelines de Kafka, sistemas de métricas como Prometheus con almacenamiento en Redis Cluster, o shards de agregación en Java.
• Bloqueos centralizados: cada región opera su propio contador localmente, reduciendo la latencia de escrituras y evitando cuellos de botella en un único punto.
• Lógica de reconciliación: el CRDT asegura convergencia automática sin escribir código de merge. Por ejemplo, en un super cluster con 20 regiones, cada una actualiza su contador local y los sources de JetStream propagan solo los deltas, no el historial completo.

En términos de rendimiento, NATS Server 2.12 maneja contadores distribuidos con:

• Latencia en el camino crítico: incrementos en milisegundos (similar a Redis local, pero sin coordinación global).
• Alto throughput: hasta 75.000 incrementos por segundo (benchmark reportado por Netflix para su solución similar, aunque NATS no publica cifras oficiales).

El impacto en la arquitectura es mínimo: solo requiere configurar el stream con AllowMsgCounter: true y AllowDirect: true. No hay que desplegar nuevos servicios, cambiar clientes existentes (el protocolo usa headers HTTP estándar) ni modificar la infraestructura de observabilidad.

Para equipos de seguridad

El mecanismo introduce vectores de ataque limitados pero relevantes:

• Inyección de deltas maliciosos: el header Nats-Incr debe cumplir con la expresión regular ^[+-]\d+$. Un mensaje con Nats-Incr: +1e100 sería rechazado por el servidor (NATS 2.12+ valida el formato estrictamente).
• Denegación de servicio por overflow: aunque se usa big.Int, un delta extremadamente grande (ej: +99999999999999999999) podría saturar recursos en el servidor. NATS limita el tamaño del cuerpo del mensaje a 1MB por defecto (configurable en max_payload), pero no hay un límite explícito para el delta.
• Suplantación de mensajes: al no requerir autenticación por subject (solo por cliente NATS), un atacante con acceso a la red podría enviar deltas falsos. La solución es integrar con los mecanismos de autenticación de NATS (como NKey o JWT) para restringir quién puede publicar en subjects de contadores.

En entornos multiinquilino, se recomienda:

1. Crear streams separados por inquilino.
2. Usar subject mappings para aislar namespaces (ej: counter.<tenant>.<region>.<metric>).
3. Aplicar políticas de ACL con JetStream authorization para restringir permisos de publicación.

Para equipos de SRE

El cambio afecta a:

• Observabilidad: los contadores distribuidos generan métricas de convergencia (tiempo hasta que todos los nodos alcanzan el mismo valor) y latencia por región. NATS Server 2.12 expone estas métricas en su endpoint de estadísticas (ej: jetstream_counter_latency).
• Escalabilidad: el flag AllowMsgCounter impone restricciones de diseño. Por ejemplo, no se pueden usar sujetos comodín (counter.*), lo que limita la flexibilidad en la nomenclatura. Para 10.000 subjects, esto requiere planificación previa.
• Resiliencia: durante una partición de red, los contadores regionales divergen temporalmente, pero convergen automáticamente al restaurarse la conectividad. Sin embargo, si un source falla y se reconfigura, el servidor mantiene un historial de los últimos valores vistos (en el header Nats-Counter-Sources) para evitar doble conteo.

Detalles técnicos

Requisitos y compatibilidad

stream:
  name: global_page_views
  subjects:
    - counter.page.*
  retention: limits
  max_msgs: -1
  max_bytes: -1
  allow_direct: true
  allow_msg_counter: true

La API de streams en NATS 2.12 introduce el campo config.allow_msg_counter (booleano), que debe establecerse en true al crear el stream. Una vez habilitado, no puede deshabilitarse.

Flujo de trabajo del contador

1. Publicación:

   nats pub counter.page.homepage --header "Nats-Incr:+1" ""
   

El servidor valida:

– Que el subject coincida con el stream (ej: counter.page.*).

– Que el header Nats-Incr cumpla con ^[+-]\d+$.

– Que el stream tenga AllowMsgCounter: true.

1. Procesamiento:

– El servidor calcula el nuevo total: nuevo_valor = valor_actual + delta.

– Almacena un nuevo mensaje con cuerpo {"val":"<nuevo_valor>"}.

– Devuelve en PubAck:

     {
       "stream":"global_page_views",
       "seq": 42,
       "val": "12345"
     }
     

1. Fuentes y agregación:

– NATS calcula el delta entre el último valor almacenado para ese source y el valor entrante.

– Actualiza el subject destino con un nuevo mensaje cuyo cuerpo es {"val":"<nuevo_total>"}.

– Mantiene un registro interno en el header Nats-Counter-Sources con pares {stream_name}#{original_subject}:{último_valor_visto}.

Limitaciones y casos de borde

• Precisión: Los valores se serializan como strings JSON para evitar problemas en clientes JavaScript. Un total de 9007199254740993 (mayor a 2^53) se representa correctamente, pero clientes que parseen el JSON como número podrían perder precisión.
• Orden de mensajes: Aunque el CRDT tolera orden arbitrario, JetStream garantiza at-least-once delivery por defecto. Si un cliente reintenta una publicación y el mensaje se duplica, el servidor aplica el delta múltiples veces. Para evitar esto, los clientes deben implementar idempotencia (ej: usar un ID único por incremento).
• Tamaño de mensajes: El cuerpo de cada mensaje es {"val":"<nuevo_total>"}. Para contadores con incrementos pequeños (ej: +1), esto es eficiente. Pero si el delta es enorme (ej: +1.000.000), el mensaje crece. NATS limita el tamaño por defecto a 1MB (max_payload), pero esto debe ajustarse según el caso de uso.

Qué deberían hacer los administradores y equipos técnicos

Actualización a NATS Server 2.12+

1. Verificar versión:

   nats-server --version
   

Debe ser 2.12.0 o superior.

1. Planificar la migración:

– Define una convención clara para los subjects de contadores (ej: counter.<dominio>.<métrica>).

1. Configurar el stream:

   nats stream add global_page_views \
     --subjects "counter.page.*" \
     --retention limits \
     --max-msgs -1 \
     --max-bytes -1 \
     --allow-direct \
     --allow-msg-counter
   

Implementar contadores en clientes

Cliente Go (usando orbit.go/counters)

1. Instalar el módulo:

   go get orbit.go/counters@latest
   

1. Crear un contador:

   import "orbit.go/counters"

   func main() {
       nc, _ := nats.Connect("nats://localhost:4222")
       defer nc.Close()

       counter, err := counters.NewCounterFromStream(nc, "global_page_views", "counter.page.homepage")
       if err != nil {
           log.Fatal(err)
       }

       // Incrementar en 1
       newVal, err := counter.Add(1)
       if err != nil {
           log.Fatal(err)
       }
       fmt.Println("Nuevo valor:", newVal)
   }
   

Cliente CLI de NATS

Para pruebas rápidas o integración con scripts:

# Incrementar en 5
nats pub counter.page.checkout --header "Nats-Incr:+5" ""

# Obtener valor (requiere AllowDirect: true)
nats req counter.page.checkout "" --header "Nats-Incr:+0"

El PubAck incluirá el valor post-incremento en el campo val.

Casos de uso prácticos

1. Contadores regionales en Kubernetes

Implementar una topología de super cluster con tres niveles:

• Regional: Cada namespace en Kubernetes tiene su propio stream con subjects como counter.<region>.<métrica>.
• Continental: Un stream agrupa los deltas regionales (ej: counter.global.<métrica>).
• Global: Un stream consolidado (ej: counter.worldwide.<métrica>).

Configuración de sources en YAML:

sources:
  - name: regional_to_continental
    stream: continental
    subject_transforms:
      - src: counter.{region}.page_views
        dest: counter.global.page_views
    filter_subject: counter.{region}.page_views

2. Rate limiting distribuido

Cada región mantiene su propio contador de tokens:

# Inicializar contador de tokens (ej: 1000 tokens)
nats pub rate.limiting.region1 --header "Nats-Incr:+1000" ""

# Consumir un token
nats pub rate.limiting.region1 --header "Nats-Incr:-1" ""

Los sources propagan los deltas, permitiendo un límite global aproximado sin bloqueos centralizados.

3. Leaderboards y billing

Para métricas que requieran consistencia eventual:

// En Go: incrementar el score de un usuario
_, err := counter.Add(1) // Ej: +100 puntos
if err != nil {
    log.Println("Error al incrementar score:", err)
}

Los clientes pueden leer el valor actual con counter.Load() y mostrarlo inmediatamente, sabiendo que convergerá a un valor consistente en todos los nodos.

Monitoreo y alertas

1. Métricas clave:

– jetstream_counter_convergence_time{stream, subject}: Tiempo hasta que todos los nodos alcanzan el mismo valor tras una partición.

– jetstream_counter_retries{stream, subject}: Cantidad de reintentos por subject.

1. Alertas:

– Si jetstream_counter_retries aumenta sin causa aparente, verificar clientes que no implementen idempotencia.

Conclusión

NATS Server 2.12 introduce un primitivo de contadores distribuidos CRDT que resuelve un problema clásico de sistemas distribuidos: contar eventos globalmente sin sacrificar latencia ni disponibilidad. Al convertir cada subject en un contador con un simple flag (AllowMsgCounter: true), los equipos eliminan la necesidad de servicios personalizados de agregación, código de reconciliación o bloqueos centralizados. La implementación usa aritmética de enteros de precisión arbitraria, tolera reintentos idempotentes y converge automáticamente gracias a las propiedades del CRDT.

Para equipos de DevOps, esto significa menos componentes en el stack y mayor resiliencia. Para infraestructura, menos latencia en el camino crítico. Y para SRE, métricas nativas de convergencia y latencia. La clave está en adoptar el patrón desde el diseño: definir una convención clara para los subjects, habilitar AllowDirect para lecturas eficientes y, sobre todo, validar que los clientes manejen reintentos de forma idempotente.

El cambio no requiere una migración compleja, pero sí planificación. Los streams existentes no pueden habilitar AllowMsgCounter, por lo que en muchos casos habrá que crear nuevos streams. Sin embargo, el ROI es inmediato: contadores globales con la misma latencia que un Redis local, sin locks ni consenso.

Fuentes

• Distributed Counter in NATS JetStream: A CRDT-Based Approach to Global Counting
• ADR-49: JetStream Distributed Counter CRDT
• Notas de lanzamiento de NATS Server 2.12
• Documentación oficial de JetStream (AllowMsgCounter)