dataforge/message_bus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b799de26d299fbb96febcfbf0e68dd4057c2e316
message_bus/reports/0006-2026-06-07-unibus-decentralization.md
T
egutierrez 29fe688b7a ahora si funciona 
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-07 16:23:53 +02:00

20 KiB
Raw Blame History

Report 0006 — unibus: descentralización / alta disponibilidad (issue 0003, fases 0003a–0003e)

  • Fecha: 07/06/2026
  • Autor: agente (Claude Opus 4.8)
  • Ámbito: projects/message_bus/apps/unibus (sub-repo dataforge/unibus). Paquetes embeddednats, busauth, membership, blobstore, client; cmd/membershipd; dev/feature_flags.json.
  • Estado: done — entrega 0003a–0003e en master del sub-repo. El despliegue multi-nodo real (0003f) lo ejecuta el humano; al final hay un runbook.
  • Origen: issue dev/issues/0003-decentralization-ha.md. Cierra además el residual H4 (ACL por subject) que el hardening (report 0005) difirió a este issue, y el agujero de anti-replay multi-nodo que la auditoría (report 0004) señaló como bloqueante antes del cluster.

Resumen

unibus deja de ser un SPOF: el servidor NATS embebido puede formar cluster (data plane replicado), el control plane y los blobs pueden vivir en JetStream KV / Object Store replicados, el cliente hace failover entre nodos, y el anti-replay pasa a un store de nonces compartido. Todo está construido por branch-by-abstraction detrás del flag decentralized (off): el comportamiento de un solo nodo (SQLite + disco) no cambia y master sigue desplegable en cada paso.

El issue marca el rollout R1→R3: con 2 nodos (magnus + homer) se despliega en R1 (réplicas=1, sin tolerancia a fallo todavía); cuando entre el tercer nodo se escala a R3 (nats stream/kv update --replicas 3) para HA real (quorum 2/3). El código soporta réplicas configurables; el despliegue elige el número.

Historia (vista git log --first-parent master):

da42051 Merge issue/0003e-client-failover: client failover + replicated nonce store + subject ACL (H4)
649dc9e Merge issue/0003d-objectstore: replicated blobs on NATS Object Store
94e7ced Merge issue/0003c-migrate-kv: idempotent SQLite->KV migration + backup
b8c9b2b Merge issue/0003b-jetstream-store: Store interface + JetStream KV backend (fail-closed)
3230b31 Merge issue/0003a-cluster: NATS cluster routes (auth + mutual TLS)

Verificación global (sin caché, en master tras los 5 merges):

$ CGO_ENABLED=0 go build ./...   # OK
$ CGO_ENABLED=0 go vet ./...     # limpio
$ CGO_ENABLED=0 go test -count=1 ./...
  ok  cmd/membershipd    0.004s
  ok  pkg/blobstore      0.096s
  ok  pkg/busauth        0.007s
  ok  pkg/client         5.857s
  ok  pkg/embeddednats   6.559s
  ok  pkg/frame          0.002s
  ok  pkg/membership     3.059s

Fase 0003a — Cluster NATS (routes con auth + TLS mutuo) ✅

Qué. pkg/embeddednats.ServerConfig gana ServerName (único por nodo, lo exige el RAFT de JetStream) y un *ClusterConfig opcional: nombre de cluster, host/puerto del listener de routes, URLs de los otros nodos, secreto de routes (Username/Password → Cluster.Authorization) y un *tls.Config de TLS mutuo. Cluster nil mantiene el servidor standalone (comportamiento legacy).

Seguridad de routes (clave, del report de auditoría 0004). Las routes son una frontera de confianza server-to-server, distinta del plano de clientes: autentican NODOS, no usuarios del bus. NO se reutiliza el authenticator nkey de clientes. busauth.RouteTLSConfig(cert, key, ca) construye el TLS mutuo: el nodo presenta su certificado firmado por la CA y verifica el del nodo entrante contra la misma CA (RequireAndVerifyClientCert), reusando la CA del issue 0001. Un nodo sin el secreto de cluster y sin un certificado firmado por la CA no puede unirse ni inyectar mensajes.

cmd/membershipd gana las flags de cluster (--cluster-name/--server-name/--cluster-port/--routes/--cluster-user/--cluster-pass/--route-tls-cert/-key/-ca). validateClusterConfig rechaza un cluster en bind público sin secreto de routes y sin TLS mutuo completo, y rechaza flags de route-TLS parciales (all-or-nothing).

Evidencia (DoD: golden + 2 edge + 2 error path):

$ CGO_ENABLED=0 go test ./pkg/embeddednats/ -run TestCluster -v
  --- PASS: TestClusterForwardsAcrossNodes   (2 nodos: subject publicado en n1 llega a un suscriptor en n0)
  --- PASS: TestClusterThreeNodesForward     (3 nodos, forma HA: publish en n2 llega a n0)
  --- PASS: TestClusterMutualTLSForwards     (forwarding sobre routes con TLS mutuo)
  --- PASS: TestClusterRejectsBadRouteAuth   (password de cluster incorrecto -> 0 routes; el cluster legítimo no cambia)
  --- PASS: TestClusterRejectsUnsignedNode   (cert no firmado por la CA -> 0 routes)
$ CGO_ENABLED=0 go test ./cmd/membershipd/ -run 'TestClusterConfigPolicy|TestSplitRoutes'   # PASS

Gaps. Los tests son in-process (varios nats-server embebidos en un proceso). El forwarding entre membershipds reales en 3 VPS + escalado a R3 es 0003f (humano). NumRoutes() cuenta conexiones del pool de routes, no peers; los tests de rechazo comparan contra un baseline estabilizado en vez de un número fijo.

Fase 0003b — Interfaz Store + jetstreamStore (KV) ✅

Qué. membership.Store pasa a interfaz (branch-by-abstraction). La implementación SQLite se renombra sqliteStore y sigue siendo el default (Open(path) la devuelve). ErrNotFound es un sentinel agnóstico del backend (la SQLite mapea sql.ErrNoRows a él; el control plane ya no importa database/sql). jetstreamStore (nuevo) implementa Store sobre cinco buckets KV replicados — rooms, members, rooms_by_member (índice inverso para ListRoomsForEndpoint), room_keys, users — con réplicas configurables (R1..R5). Fail-closed: cada lectura está acotada por OpTimeout e IsAuthorized/HasAdmin devuelven false ante cualquier error de backend (una pérdida de quorum del KV deniega, nunca admite), exactamente lo que la auditoría exigía.

Flag decentralized (off) añadido a dev/feature_flags.json.

Evidencia (DoD: golden + edges + error path):

$ CGO_ENABLED=0 go test ./pkg/membership/ -run TestJetStreamStore -v
  --- PASS: TestJetStreamStoreRoomsCRUD              (room cifrada + owner + invitado; latest-epoch + rekey)
  --- PASS: TestJetStreamStoreUsers                  (add/get/authorize/list/revoke + gate admin, normalización case-insensitive, duplicado rechazado)
  --- PASS: TestJetStreamStoreNotFound               (mapeo ErrNotFound en misses)
  --- PASS: TestJetStreamStoreIsAuthorizedFailClosed (NATS apagado -> IsAuthorized y HasAdmin DENIEGAN dentro del timeout)

Gaps. KV no tiene transacción multi-key: CreateRoom/AddMember son secuencias de Puts idempotentes con orden recoverable; documentado en el código. El wiring de arranque de membershipd-en-KV (selección del store según flag) se difiere a 0003e/0003f: tiene un ciclo bootstrap (la conexión interna de membershipd al NATS embebido debería autenticarse contra el mismo store que aún no existe), que pertenece al rediseño de sesión/deploy. El jetstreamStore queda completo, testeado y consumido inmediatamente por migrate-to-kv (0003c).

Fase 0003c — migrate-to-kv (idempotente, con backup) ✅

Qué. Snapshot/SealedKeyRecord (volcado agnóstico del control plane: rooms con su epoch real, members, todas las filas de sealed-key por epoch, users con status). ExportSnapshot en ambos backends; importSnapshot del KV escribe con Puts crudos (conserva epoch/status, no resetea a defaults), así que la migración es fiel e idempotente (cada escritura es un overwrite → re-ejecutar converge). MigrateSQLiteToKV orquesta export→import; BackupSQLite hace una copia consistente con VACUUM INTO antes de migrar. Comando membershipd migrate-to-kv --db <path> --nats-url <url> [--replicas N] [--ca <cert>] [--no-backup].

Evidencia (DoD: paridad + idempotencia + backup) + smoke del binario:

$ CGO_ENABLED=0 go test ./pkg/membership/ -run 'TestMigrate|TestBackup' -v
  --- PASS: TestMigrateSQLiteToKVParity            (2 rooms una rekeyed a epoch 2 + user revocado; KV == SQLite tras migrar)
  --- PASS: TestMigrateSQLiteToKVIdempotent        (migrar dos veces -> mismo estado KV)
  --- PASS: TestBackupSQLiteCreatesConsistentCopy  (el backup reabre con los mismos datos)

$ # smoke del binario (seed user -> server -> migrate -> re-run)
$ membershipd user add --handle alice --sign-pub <64hex> --role admin --db unibus.db
$ membershipd --bind 127.0.0.1 --http-port 18888 --nats-port 14888 --db unibus.db ... &
$ membershipd migrate-to-kv --db unibus.db --nats-url nats://127.0.0.1:14888 --replicas 1
  backed up unibus.db -> unibus.db.bak.<ts>
  migrated to KV (replicas=1): 0 rooms, 0 members, 0 keys, 1 users
$ membershipd migrate-to-kv ... --no-backup     # re-run idempotente: 1 users, igual

Gaps. El comando conecta al NATS con --nats-url (+ --ca opcional para TLS); la autenticación de la conexión interna con nkey de nodo en un cluster enforce se afina en el deploy 0003f.

Fase 0003d — Blobs replicados (NATS Object Store) ✅

Qué. blobstore.Store pasa a interfaz. El backend de disco se renombra diskStore (default). objectStore (nuevo) implementa Store sobre un bucket NATS Object Store con réplicas configurables (R1..R5). El direccionamiento por contenido (sha256-hex) es idéntico, así que el contrato de wire no cambia y un BlobRef es portable entre backends. membership.Server.blobs/NewServer toman la interfaz.

Evidencia (DoD: golden + edge + contrato):

$ CGO_ENABLED=0 go test ./pkg/blobstore/ -v
  --- PASS: TestObjectStoreRoundTrip           (put/get/has + dedup content-addressed)
  --- PASS: TestObjectStoreMissing             (hash desconocido ausente e ilegible)
  --- PASS: TestObjectStoreAddressMatchesDisk  (Object Store y disco direccionan los mismos bytes al MISMO hash)

Gaps. Como el KV store, la selección del Object Store en membershipd se difiere al boot path decentralized; disco sigue default. Sin cuota/GC (eso es el issue 0002 / H9).

Fase 0003e — Cliente failover + nonce replicado + ACL por subject (H4) ✅

Tres componentes, tres commits atómicos en la rama.

1. Failover de cliente

Options.NatsServers (seeds extra) y Options.CtrlURLs (control planes extra). El cliente conecta a la lista unida con MaxReconnects(-1) + RetryOnFailedConnect, así nats.go hace failover a un nodo vivo cuando cae el que tenía. doJSON/putBlob/getBlob prueban cada control plane en orden (failover de transporte; cada intento firma con nonce fresco, nunca es replay). New/Connect/NewWithOptions mantienen su firma (worker/chat/mobile/playground intactos). ConnectedServer()/IsConnected() para observabilidad.

$ CGO_ENABLED=0 go test ./pkg/client/ -run TestClientFailoverAcrossNodes -v
  --- PASS  (A suscrito en n0 recibe; se MATA el nodo de A; A reconecta al superviviente y SIGUE recibiendo — sesión intacta)

2. Anti-replay replicado (nonce store en KV)

El nonceCache por-proceso rompe el anti-replay en multi-nodo: un request capturado y reenviado a OTRO nodo (cuyo cache nunca vio el nonce) se aceptaría. nonceStore pasa a interfaz; memNonceCache (default) + kvNonceStore nuevo que reclama cada nonce con un Create atómico en un bucket compartido (primer uso gana; cualquier uso posterior en cualquier nodo rechaza). Error de backend → fail closed (rechaza). TTL del bucket = nonceTTL (2·clockSkew). Server.UseReplicatedNonces(js, replicas) lo activa por nodo.

$ CGO_ENABLED=0 go test ./pkg/membership/ -run TestReplicatedNonceRejectsCrossNodeReplay -v
  --- PASS  (request aceptada (200) en nodo A; mismo ts+nonce reenviado a nodo B -> 401 replayed; replay a A de nuevo -> 401)

3. ACL por subject derivada de pertenencia (residual H4)

busauth.NewNkeyAuthenticatorACL + PermissionsFunc: el authenticator, tras autorizar, deriva y RegisterUser()a permisos por subject; error de derivación → deniega (fail closed). membership.SubjectACLFor(store) mapea una identidad a los subjects de sus rooms + la infraestructura de cliente (_INBOX.>, $JS.API.>). client.RefreshSession() reconecta el data plane para que el authenticator re-derive permisos tras un cambio de membresía (NATS congela permisos al conectar).

$ CGO_ENABLED=0 go test ./pkg/membership/ -run 'TestSubjectACLIsolation|TestRefreshSessionGainsNewRoom' -v
  --- PASS: TestSubjectACLIsolation        (alice miembro de room.A: sub/pub room.A OK; sub y pub room.B -> permissions violation;
                                            alice nunca lee el tráfico de bob en room.B; bob nunca recibe el publish cruzado de alice)
  --- PASS: TestRefreshSessionGainsNewRoom (alice sin permiso para room B hasta que se la añade y llama RefreshSession; el reconnect
                                            le concede el subject y entonces recibe el tráfico de room B)

Estado del residual H4. El aislamiento por subject derivado de pertenencia está implementado y demostrado a nivel de data plane: un peer registrado ya no puede sub/pub en subjects arbitrarios, queda confinado a las rooms a las que pertenece, y el contenido de las demás (ya E2E desde el hardening) le es invisible incluso como metadata de subject. El authenticator ACL es opt-in (NewServer/membershipd mantienen el authenticator abierto por defecto) y se cablea con el boot path decentralized. Lo que queda para 0003f/futuro: el refresco automático y transparente de permisos en cada cambio de membresía — hoy RefreshSession es una llamada explícita que el peer hace tras unirse a una room (porque NATS congela permisos al conectar, el flujo "connect → create → publish en la misma conexión" requiere un refresh entre el create y el publish). Esto es deliberado para no romper el flujo legacy ni los tests existentes; se activa con el rediseño de sesión del deploy.

Cobertura de tests (DoD)

Fase Tests nuevos Tipo
0003a TestCluster* (5) + TestClusterConfigPolicy + TestSplitRoutes golden + 2 edge + 2 error path
0003b TestJetStreamStore* (4) golden + edges + fail-closed
0003c TestMigrate* (2) + TestBackup* + smoke binario paridad + idempotencia + backup
0003d TestObjectStore* (3) golden + edge + contrato
0003e TestClientFailoverAcrossNodes, TestReplicatedNonceRejectsCrossNodeReplay, TestSubjectACLIsolation, TestRefreshSessionGainsNewRoom failover + replay + aislamiento + refresh

Todo el suite previo sigue verde: el comportamiento de un solo nodo (SQLite + disco, authenticator abierto, nonce en memoria) no cambia.

Resumen para 0003f (despliegue multi-nodo — lo ejecuta el humano)

Pre-requisitos del issue: alta del alias SSH + clave de cada nodo, integración en la WireGuard, y revisar/aligerar la carga existente (magnus corre coolify/minio/postgres/etc.). Nodos hoy: magnus y homer (141.94.69.66); falta el tercero para quorum real.

Despliegue R1 (2 nodos, magnus + homer — funciona, sin tolerancia a fallo todavía)

  1. CA y certificados de nodo. Generar (o reusar) la CA del 0001 (deploy/tls/ca.crt + ca.key custodiada fuera de los hosts de aplicación, en pass). Emitir un certificado de route por nodo cuyo SAN cubra la IP/host de su puerto de cluster, firmado por esa CA (sirve para server y client auth en el handshake mutuo de routes).
  2. Secreto de cluster. Un --cluster-user/--cluster-pass compartido por los dos nodos (en pass).
  3. Arrancar cada membershipd con bind público + --bus-auth enforce + TLS del data plane (0001) + cluster: 
    membershipd --bind <ip> --bus-auth enforce \
  --tls-cert deploy/tls/server.crt --tls-key deploy/tls/server.key \
  --cluster-name unibus --server-name <magnus|homer> --cluster-port 6250 \
  --cluster-user "$CL_USER" --cluster-pass "$CL_PASS" \
  --route-tls-cert deploy/tls/<node>.crt --route-tls-key deploy/tls/<node>.key --route-tls-ca deploy/tls/ca.crt \
  --routes nats://$CL_USER:$CL_PASS@<otro-nodo>:6250
    validateClusterConfig exige el secreto + TLS mutuo completo en bind público (rechaza arrancar inseguro).
  4. systemd Restart=always en cada nodo (un SIGTERM limpio es exit 0; on-failure no reiniciaría — gotcha conocido). Reusar deploy/unibus-membershipd.service + deploy/install.sh.
  5. Streams/KV/Object Store en R1 mientras solo haya 2 nodos (réplicas=1): el data plane efímero (core-NATS) ya tolera la caída de uno (los clientes reconectan al otro); el control plane KV y las rooms persistentes NO toleran caída hasta R3.
  6. Clientes: pasarles la lista de los 2 endpoints — Options{NatsServers: [n2], CtrlURLs: [ctrl2]} además del primario — para que el failover de cliente (0003e) funcione.

Nota sobre el flag decentralized: en R1 con 2 nodos se puede operar el control plane aún en SQLite por-nodo (cada membershipd su BD) si no se quiere KV todavía; para que cualquier nodo sirva cualquier request hay que activar el control plane KV (decentralized: on) — eso requiere completar el wiring de arranque KV (selección del store + bootstrap del authenticator interno) que esta entrega dejó documentado como el paso de integración de 0003f. La migración de los datos existentes la hace membershipd migrate-to-kv (probado) antes de activar el flag.

Escalado a R3 (cuando entre el tercer nodo — HA real)

  1. Preparar el tercer nodo (SSH + WireGuard + cert de route firmado por la CA).
  2. Añadirlo al cluster (su --routes apunta a magnus+homer y viceversa; mismo --cluster-name y secreto).
  3. Escalar en caliente, sin downtime ni migración de datos: 
    nats stream update <stream> --replicas 3
nats kv update <bucket> --replicas 3      # rooms/members/rooms_by_member/room_keys/users/nonces
nats obj update UNIBUS_blobs --replicas 3
    o crear los buckets con Replicas: 3 desde el inicio en el tercer nodo (el código ya lo soporta vía JetStreamConfig.Replicas / ObjectStoreConfig.Replicas).
  4. Activar decentralized: on en dev/feature_flags.json (con el wiring de arranque KV completo).
  5. Chaos test real (DoD del issue): matar 1 de 3 nodos → el control plane sigue aceptando escrituras (quorum 2/3); matar 2 de 3 → las escrituras se bloquean (quorum perdido, comportamiento esperado y documentado, no corrupción).

No olvidar en 0003f

  • Nonce replicado obligatorio antes de exponer el cluster: llamar Server.UseReplicatedNonces(js, replicas) en cada nodo. Sin esto el anti-replay es nulo en multi-nodo (un replay a otro nodo se acepta). Probado en TestReplicatedNonceRejectsCrossNodeReplay.
  • ACL por subject: si se quiere el aislamiento de data plane en producción, usar busauth.NewNkeyAuthenticatorACL(store.IsAuthorized, perms) con perms derivado de membership.SubjectACLFor(store), y que los clientes llamen RefreshSession() tras unirse a una room. Si no se activa, el residual sigue siendo: un peer registrado observa metadata de tráfico (no contenido) de subjects que conozca y puede spamear bytes descartados — irrelevante en un despliegue solo-WireGuard.

Gaps / pendientes honestos de este trabajo

  • No hay despliegue real ni chaos test sobre 3 VPS — todo se valida in-process (varios nats-server embebidos en un proceso). El chaos test real (matar 1/3, matar 2/3) es 0003f.
  • El wiring de arranque membershipd-en-decentralized no se cableó (selección del store/blobstore KV + bootstrap del authenticator interno). El jetstreamStore, el objectStore, el kvNonceStore y el authenticator ACL están completos y testeados, pero el binario solo los expone como API; conectarlos en main.go (con el flag) pertenece a la integración de 0003f por el ciclo bootstrap descrito. Es branch-by-abstraction correcto (flag off, master verde), pero significa que --decentralized no es todavía un interruptor del binario.
  • KV no es transaccional: CreateRoom/AddMember son secuencias de Puts idempotentes; una caída a mitad deja un estado parcial recoverable (no corrupto), no atómico. Aceptable para el control plane (baja frecuencia, owner-serializado).
  • El refresco de ACL no es transparente: RefreshSession es una llamada explícita; el auto-refresh en cada cambio de membresía (rehaciendo suscripciones) queda para el rediseño de sesión.
  • No se ejecutó govulncheck sobre nats-server/nats.go/modernc (paso de CI aparte, ya anotado en el report 0005).
Reference in New Issue View Git Blame Copy Permalink