From b7ccf9103bdf66f79cda56504d44d9eb7682b8b8 Mon Sep 17 00:00:00 2001 From: AlpNuhoglu Date: Tue, 16 Jun 2026 16:57:54 +0300 Subject: [PATCH] feat(messaging): add NATS JetStream event bus with durable consumers - introduce NATSBus alongside existing RedisBus - preserve Publisher/Subscriber abstractions - add durable JetStream streams and consumers - implement ACK-after-success processing model - support event replay and redelivery - preserve OpenTelemetry trace propagation across async boundaries - add transport factory with EVENT_BUS selection - add event bus metrics and messaging documentation - keep Redis as default transport for backward compatibility --- .env.example | 11 + cmd/leaderboard/main.go | 10 +- cmd/matchmaking/main.go | 10 +- cmd/websocket/main.go | 11 +- docker-compose.yml | 33 +++ docs/messaging.md | 307 +++++++++++++++++++++ go.mod | 12 +- go.sum | 25 +- internal/wsgateway/bridge.go | 28 +- pkg/config/config.go | 13 + pkg/events/events.go | 24 ++ pkg/events/events_test.go | 4 +- pkg/events/factory.go | 54 ++++ pkg/events/nats.go | 382 +++++++++++++++++++++++++++ pkg/events/nats_test.go | 146 ++++++++++ pkg/events/redis.go | 22 +- pkg/events/trace_propagation_test.go | 2 +- pkg/metrics/metrics.go | 31 +++ tests/integration/redis_test.go | 2 +- 19 files changed, 1105 insertions(+), 22 deletions(-) create mode 100644 docs/messaging.md create mode 100644 pkg/events/factory.go create mode 100644 pkg/events/nats.go create mode 100644 pkg/events/nats_test.go diff --git a/.env.example b/.env.example index c762c78..4b1c9f6 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,17 @@ REDIS_ADDR=redis:6379 REDIS_PASSWORD= REDIS_DB=0 +# --- Event transport --- +# Which messaging transport carries inter-service events. Redis remains the +# system of record (sessions, matchmaking queue, leaderboard, room cache) +# regardless of this setting; EVENT_BUS only swaps the event-delivery layer. +# nats -> NATS JetStream: durable, replayable, at-least-once (default in compose) +# redis -> Redis Pub/Sub: fire-and-forget, at-most-once (original behavior) +EVENT_BUS=nats +NATS_URL=nats://nats:4222 +# Bounded worker pool size for the JetStream consumers (per service). +EVENT_WORKERS=8 + # --- Service URLs (as seen from the API gateway) --- PLAYER_SERVICE_URL=http://player:8081 MATCHMAKING_SERVICE_URL=http://matchmaking:8082 diff --git a/cmd/leaderboard/main.go b/cmd/leaderboard/main.go index 43a08c6..62bc08f 100644 --- a/cmd/leaderboard/main.go +++ b/cmd/leaderboard/main.go @@ -43,7 +43,15 @@ func main() { } m := metrics.New(cfg.ServiceName) - bus := events.NewRedisBus(rdb, log) + bus, err := events.NewBus(events.Config{ + Transport: cfg.EventBus, + DurableName: cfg.ServiceName, + Workers: cfg.EventWorkers, + }, rdb, cfg.NATSURL, m, log) + if err != nil { + log.Fatal("failed to init event bus", zap.Error(err)) + } + defer func() { _ = bus.Close() }() svc := leaderboard.NewService(leaderboard.NewStore(rdb), bus, m, log) handler := leaderboard.NewHandler(svc) diff --git a/cmd/matchmaking/main.go b/cmd/matchmaking/main.go index a5461e8..285591a 100644 --- a/cmd/matchmaking/main.go +++ b/cmd/matchmaking/main.go @@ -43,7 +43,15 @@ func main() { } m := metrics.New(cfg.ServiceName) - bus := events.NewRedisBus(rdb, log) + bus, err := events.NewBus(events.Config{ + Transport: cfg.EventBus, + DurableName: cfg.ServiceName, + Workers: cfg.EventWorkers, + }, rdb, cfg.NATSURL, m, log) + if err != nil { + log.Fatal("failed to init event bus", zap.Error(err)) + } + defer func() { _ = bus.Close() }() svc := matchmaking.NewService( matchmaking.NewQueue(rdb), matchmaking.NewRoomStore(rdb, cfg.RoomTTL), diff --git a/cmd/websocket/main.go b/cmd/websocket/main.go index 49e3fdf..e3cebbc 100644 --- a/cmd/websocket/main.go +++ b/cmd/websocket/main.go @@ -51,7 +51,16 @@ func main() { ctx, stop := server.ShutdownContext() defer stop() - bridge := wsgateway.NewBridge(hub, events.NewRedisBus(rdb, log), log) + bus, err := events.NewBus(events.Config{ + Transport: cfg.EventBus, + DurableName: cfg.ServiceName, + Workers: cfg.EventWorkers, + }, rdb, cfg.NATSURL, m, log) + if err != nil { + log.Fatal("failed to init event bus", zap.Error(err)) + } + defer func() { _ = bus.Close() }() + bridge := wsgateway.NewBridge(hub, bus, log) go func() { if err := bridge.Run(ctx); err != nil { log.Fatal("event bridge failed", zap.Error(err)) diff --git a/docker-compose.yml b/docker-compose.yml index 496e422..5f4876b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -10,6 +10,13 @@ x-service-env: &service-env REDIS_ADDR: redis:6379 REDIS_PASSWORD: ${REDIS_PASSWORD:-} ALLOWED_ORIGINS: ${ALLOWED_ORIGINS:-*} + # Event transport. Redis stays the system of record (sessions, queue, + # leaderboard, room cache); NATS JetStream carries inter-service events with + # durability, replay and at-least-once delivery. Set EVENT_BUS=redis to fall + # back to the original Pub/Sub transport — no code change, just config. + EVENT_BUS: ${EVENT_BUS:-nats} + NATS_URL: ${NATS_URL:-nats://nats:4222} + EVENT_WORKERS: ${EVENT_WORKERS:-8} # Distributed tracing: every service ships spans to the OTel Collector, which # forwards them to Jaeger. Set OTEL_ENABLED=false to turn tracing off entirely. OTEL_ENABLED: ${OTEL_ENABLED:-true} @@ -48,6 +55,25 @@ services: timeout: 3s retries: 10 + # NATS with JetStream enabled (-js): the durable event transport. Streams are + # created idempotently by the services on boot. Storage is persisted to a + # volume so events (and consumer cursors) survive container restarts — + # enabling replay and at-least-once redelivery. :8222 serves the monitoring + # endpoint used for the healthcheck. + nats: + image: nats:2.10-alpine + command: ["-js", "-sd", "/data", "-m", "8222"] + ports: + - "4222:4222" # client + - "8222:8222" # monitoring + volumes: + - natsdata:/data + healthcheck: + test: ["CMD", "wget", "-qO-", "http://localhost:8222/healthz"] + interval: 5s + timeout: 3s + retries: 10 + gateway: build: context: . @@ -113,6 +139,8 @@ services: depends_on: redis: condition: service_healthy + nats: + condition: service_healthy otel-collector: condition: service_started healthcheck: @@ -129,6 +157,8 @@ services: depends_on: redis: condition: service_healthy + nats: + condition: service_healthy otel-collector: condition: service_started healthcheck: @@ -147,6 +177,8 @@ services: depends_on: redis: condition: service_healthy + nats: + condition: service_healthy otel-collector: condition: service_started healthcheck: @@ -209,3 +241,4 @@ services: volumes: pgdata: + natsdata: diff --git a/docs/messaging.md b/docs/messaging.md new file mode 100644 index 0000000..7e12388 --- /dev/null +++ b/docs/messaging.md @@ -0,0 +1,307 @@ +# Messaging & Event Transport + +GameMesh is event-driven: services communicate asynchronously through a small, +transport-agnostic event contract. This document explains the migration of that +transport from **Redis Pub/Sub** to **NATS JetStream**, the guarantees each +provides, and the path toward a transactional Outbox. + +--- + +## 1. Why migrate from Redis Pub/Sub + +Redis Pub/Sub was the original transport. It is fast and was already in the +stack, but it is **fire-and-forget**: + +- A subscriber that is **down or restarting** when a message is published + **never sees it** — the message is dropped at the broker. +- There is **no acknowledgement** — the publisher cannot know a consumer + processed the event, and a consumer cannot ask for redelivery on failure. +- There is **no history / replay** — you cannot re-read past events to rebuild + state or onboard a new consumer. + +For realtime hints (a leaderboard tick will repair itself on the next poll) this +is acceptable. For events that drive state and user-visible actions +(`MatchFound` routes players into a game), losing a message is a correctness +problem. NATS JetStream gives us **durability, acknowledgements, replay and +at-least-once delivery** while keeping the same clean abstraction. + +> **Redis is not removed.** It remains the system of record for sessions, the +> matchmaking queue, the leaderboard sorted set, and the room cache. JetStream +> replaces **only** the inter-service event transport. + +--- + +## 2. Redis Pub/Sub vs NATS JetStream + +| Property | Redis Pub/Sub | NATS JetStream | +| --------------------- | ---------------------------- | ------------------------------------ | +| Delivery guarantee | At-most-once | **At-least-once** | +| Durability | None (in-memory fan-out) | **Persisted to a stream** (file) | +| Consumer offline | Messages lost | **Buffered, delivered on reconnect** | +| Acknowledgement | None | **Explicit ACK / NAK / TERM** | +| Retry on failure | Not possible | **Automatic redelivery (MaxDeliver)**| +| Replay / history | Not possible | **DeliverAll / by time / by seq** | +| Backpressure | None (drops) | **Flow control (pending limits)** | +| Outbox compatibility | Poor (no durable hand-off) | **Natural durable hand-off point** | +| Latency | Sub-millisecond | Low (single-digit ms, persisted) | + +Core NATS (without JetStream) was rejected because it shares Pub/Sub's +at-most-once / no-replay limitations. JetStream is required for the durability +and Outbox-readiness goals. + +--- + +## 3. Delivery guarantees + +The architectural seam is unchanged: services depend only on +[`events.Publisher`](../pkg/events/events.go) and +[`events.Subscriber`](../pkg/events/events.go). Two transports implement them: + +- `RedisBus` — at-most-once (unchanged, kept for fallback / comparison). +- `NATSBus` — at-least-once. + +### How at-least-once is enforced without changing the interface + +The legacy `Subscriber` returns a channel and has **no place to signal +processing success**. Rather than break that interface (and every service using +it), we added an **opt-in** richer contract: + +```go +type Handler func(ctx context.Context, e Event) error + +type AckSubscriber interface { + SubscribeAck(ctx context.Context, handler Handler, topics ...string) error +} +``` + +`NATSBus` implements **both** `Subscriber` (channel, drop-in compatible) and +`AckSubscriber` (handler, true at-least-once). The WS bridge type-asserts: + +```go +if ack, ok := sub.(events.AckSubscriber); ok { + return ack.SubscribeAck(ctx, b.dispatch, topics...) // ACK after success +} +// else: legacy channel path +``` + +A message is **ACK'd only after the handler returns `nil`**. On error it is +**NAK'd with a delay** and redelivered up to `MaxDeliver` (5). Malformed/poison +messages are **TERM'd** (never redelivered). Because redelivery is possible, +**handlers must be idempotent**. + +--- + +## 4. Replay examples + +JetStream persists events, so they can be re-read. Using the `nats` CLI: + +```bash +# Inspect a stream and its consumers +nats stream info MATCHMAKING +nats consumer info MATCHMAKING websocket-matchmaking + +# Tail everything currently in a stream (no ACK side effects) +nats stream view MATCHMAKING + +# Replay the full history into a brand-new ephemeral consumer +nats consumer add MATCHMAKING replay-demo \ + --filter 'events.matchmaking.>' --ephemeral \ + --deliver all --ack none +nats consumer next MATCHMAKING replay-demo --count 100 + +# Replay only events from the last 10 minutes +nats consumer add MATCHMAKING replay-recent --ephemeral \ + --deliver 'by_start_time' --start-time '10m ago' --ack none +``` + +In code, replay is simply a consumer created with `DeliverPolicy: +DeliverAllPolicy` — which is exactly what a freshly-started service gets on its +first boot, so a new consumer automatically catches up on history. This is +verified by `TestNATSReplay` in +[`pkg/events/nats_test.go`](../pkg/events/nats_test.go): an event published +**before** any consumer exists is still delivered. + +--- + +## 5. Stream topology + +Streams are **per-domain, not a single catch-all**, so each domain's retention, +limits, storage and (future) replication can be tuned independently and one +domain's load never affects another. + +| Stream | Subjects | Retention (MaxAge) | Rationale | +| ------------- | ------------------------------------------------- | ------------------ | ----------------------------------------------------- | +| `MATCHMAKING` | `events.matchmaking`, `events.matchmaking.>` | 24h | Match events are valuable to replay for a day. | +| `LEADERBOARD` | `events.leaderboard`, `events.leaderboard.>` | 1h | Score updates are high-volume and quickly disposable. | + +Storage is `FileStorage` (persisted to the `natsdata` volume) with +`LimitsPolicy` retention. + +### Subject naming + +The publish **topics** keep their existing constant values +(`events.matchmaking`, `events.leaderboard`) so the **event contract is +unchanged**. The actual publish subject appends the event type: + +``` +events.matchmaking.MatchFound +events.leaderboard.LeaderboardUpdated +``` + +The stream's hierarchical wildcard (`events.matchmaking.>`) captures these and +leaves room for future subtypes (`events.matchmaking.MatchCancelled`) **without +new streams**, and lets future consumers filter per-type if needed. + +--- + +## 6. Consumer topology + +| Consumer (durable name) | Stream | Filter | Owner | +| ---------------------------- | ------------- | --------------------------- | ------------ | +| `websocket-matchmaking` | `MATCHMAKING` | `events.matchmaking.>` | WS bridge | +| `websocket-leaderboard` | `LEADERBOARD` | `events.leaderboard.>` | WS bridge | + +Each consumer is a **durable pull consumer**: + +- **Durable** — named after the consuming service, so restarts resume from the + last ACK'd position (no replay storm, no loss). +- `AckExplicitPolicy` — manual ACK required. +- `AckWait: 30s` — if not ACK'd in time, redeliver. +- `MaxDeliver: 5` — bounded retries, then stop (future: dead-letter subject). +- `DeliverAllPolicy` — a fresh consumer replays history (catch-up on first boot). + +### Concurrency model (no goroutine-per-message) + +A goroutine per message is unbounded and risks OOM under burst. Instead each +consumer runs a **bounded worker pool** (`EVENT_WORKERS`, default 8): + +``` +JetStream pull ──> internal work channel ──> [worker 1 .. worker N] ──> handler ──> ACK/NAK +``` + +- **Backpressure**: `PullMaxMessages` caps in-flight (unacked) messages; the + pull side blocks when the pool is saturated, so consumers never outrun + processing. +- **Graceful shutdown**: on `ctx` cancellation the consume loop stops pulling, + the work channel is closed, workers drain buffered messages, then the NATS + connection is `Drain()`ed. In-flight, un-ACK'd messages simply redeliver + later — safe because handlers are idempotent. +- **Context cancellation** propagates from the service's shutdown context all + the way into the pool. + +--- + +## 7. Trace propagation across the async boundary + +Distributed tracing (OpenTelemetry → OTel Collector → Jaeger) is preserved +**unchanged in mechanism**. The `Event` envelope carries a `Carrier +map[string]string` holding W3C `traceparent`/`tracestate`: + +``` +matchmaking.tick ──(publish: inject Carrier)──> JetStream ──(consume: extract Carrier)──> ws.dispatch +``` + +- **Publish** starts a `SpanKindProducer` span and injects the active trace + context into `Event.Carrier` via the OTel propagator. +- **Consume** extracts the Carrier and starts a `SpanKindConsumer` span + (`events.consume `), then the bridge's `ws.dispatch` runs under it. + +Because this relies only on the propagator and the Carrier field, it is +**transport-agnostic** — Redis and NATS use the identical inject/extract calls; +only the `messaging.system` span attribute differs (`redis` vs `nats`). A single +`MatchFound` therefore remains **one connected trace** in Jaeger: +`matchmaking.tick → events.publish → events.consume → ws.dispatch`. Verified by +`TestNATSPublishConsumeAck`. + +--- + +## 8. Metrics + +Both transports emit the same Prometheus instruments (registered per service in +[`pkg/metrics`](../pkg/metrics/metrics.go)), so dashboards work regardless of +`EVENT_BUS`: + +| Metric | Labels | Meaning | +| ---------------------------------------------- | --------------- | ------------------------------------------ | +| `gamemesh_events_published_total` | `topic`, `type` | Events published. | +| `gamemesh_events_consumed_total` | `topic`, `type` | Events processed and ACK'd. | +| `gamemesh_events_failed_total` | `topic`, `stage`| Failures by stage (`publish`/`decode`/`handle`). | +| `gamemesh_event_processing_duration_seconds` | `topic`, `type` | Handler latency histogram. | + +--- + +## 9. Configuration + +| Variable | Default | Purpose | +| --------------- | -------------------- | ------------------------------------------ | +| `EVENT_BUS` | `redis` (app) / `nats` (compose) | Transport selection: `redis` or `nats`. | +| `NATS_URL` | `nats://localhost:4222` | NATS endpoint (compose: `nats://nats:4222`). | +| `EVENT_WORKERS` | `8` | Bounded worker-pool size per consumer. | + +The transport switch lives in **one place**, +[`events.NewBus`](../pkg/events/factory.go), so adding a future transport +(Kafka, gRPC streaming) touches a single file and **no service code**. + +--- + +## 10. Future Outbox integration strategy + +The current design is deliberately **Outbox-ready** but does **not** implement +the Outbox pattern yet. The seam that makes it cheap: producers depend only on +`events.Publisher`. + +Today (direct publish): + +``` +service ──> Publisher.Publish ──> JetStream +``` + +A dual-write risk exists: the DB commit and the publish are separate operations, +so a crash between them can lose or duplicate an event. The Outbox pattern fixes +this **without touching producer code**: + +1. **`OutboxPublisher implements events.Publisher`** — instead of publishing to + NATS, `Publish` inserts the event row into an `outbox` table **inside the same + DB transaction** as the business state change. Atomic: either both commit or + neither does. +2. **A relay process** polls the `outbox` table (or tails the WAL via CDC) and + calls the real `NATSBus.Publish` for each unsent row, marking it sent on + success. JetStream's publish ACK confirms durable hand-off; the relay retries + on failure. + +``` +service ──> OutboxPublisher.Publish ──> [DB tx: state + outbox row] + │ + relay (poll/CDC) ─┘──> NATSBus.Publish ──> JetStream +``` + +Because `OutboxPublisher` satisfies the **same interface**, wiring it is a +factory change (`EVENT_BUS=outbox`) — **zero changes** to matchmaking, +leaderboard, or any business logic. JetStream being durable makes it the natural +sink for the relay, and at-least-once + idempotent handlers already tolerate the +duplicates an Outbox relay can produce. + +--- + +## 11. Verification + +```bash +go mod tidy +go build ./... +go test ./... # includes embedded-JetStream tests in pkg/events +docker compose up # brings up nats (JetStream), redis, jaeger, etc. +``` + +Embedded-server tests (no docker needed) in +[`pkg/events/nats_test.go`](../pkg/events/nats_test.go) prove: + +- `TestNATSPublishConsumeAck` — publish → consume → ACK with trace continuity. +- `TestNATSRedeliversOnHandlerError` — at-least-once: failed handler → NAK → + redelivery. +- `TestNATSReplay` — an event published before any consumer existed is still + delivered (replay). + +End-to-end with the stack running: generate a `MatchFound` (queue two players in +matchmaking) and a `LeaderboardUpdated` (submit a score), then confirm the event +reaches the WS client, `nats consumer info` shows it ACK'd, Jaeger shows one +connected trace, and a replay consumer re-reads it. diff --git a/go.mod b/go.mod index 02a6e95..8f51fe0 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,8 @@ require ( github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 github.com/jackc/pgx/v5 v5.9.0 + github.com/nats-io/nats-server/v2 v2.14.2 + github.com/nats-io/nats.go v1.52.0 github.com/prometheus/client_golang v1.20.5 github.com/redis/go-redis/extra/redisotel/v9 v9.7.3 github.com/redis/go-redis/v9 v9.7.3 @@ -25,7 +27,7 @@ require ( go.opentelemetry.io/otel/trace v1.44.0 go.uber.org/zap v1.27.0 golang.org/x/crypto v0.52.0 - golang.org/x/time v0.12.0 + golang.org/x/time v0.15.0 gorm.io/driver/postgres v1.5.11 gorm.io/gorm v1.30.0 ) @@ -35,6 +37,7 @@ require ( github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect github.com/Microsoft/go-winio v0.6.2 // indirect github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 // indirect + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/bytedance/gopkg v0.1.4 // indirect github.com/bytedance/sonic v1.15.1 // indirect @@ -65,6 +68,7 @@ require ( github.com/go-playground/validator/v10 v10.30.2 // indirect github.com/goccy/go-json v0.10.6 // indirect github.com/goccy/go-yaml v1.19.2 // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect @@ -72,13 +76,14 @@ require ( github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.18.6 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/mattn/go-isatty v0.0.22 // indirect github.com/mdelapenya/tlscert v0.2.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/go-archive v0.2.0 // indirect github.com/moby/moby/api v1.54.1 // indirect @@ -91,6 +96,9 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/nats-io/jwt/v2 v2.8.2 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pelletier/go-toml/v2 v2.3.1 // indirect diff --git a/go.sum b/go.sum index f038744..c23ff7b 100644 --- a/go.sum +++ b/go.sum @@ -10,6 +10,8 @@ github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 h1:uvdUDbHQHO github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc= github.com/alicebob/miniredis/v2 v2.34.0 h1:mBFWMaJSNL9RwdGRyEDoAAv8OQc5UlEhLDQggTglU/0= github.com/alicebob/miniredis/v2 v2.34.0/go.mod h1:kWShP4b58T1CW0Y5dViCd5ztzrDqRWqM3nksiyXk5s8= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= @@ -89,6 +91,8 @@ github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= @@ -110,8 +114,8 @@ github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -132,6 +136,8 @@ github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= @@ -157,6 +163,16 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc= +github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc= +github.com/nats-io/nats-server/v2 v2.14.2 h1:Q7dRhCY03Y00rETFW3KV+KGaCIajlDfWgWUVgbMxyuk= +github.com/nats-io/nats-server/v2 v2.14.2/go.mod h1:lWpb1bSpRELZfRdlMkdz8E7lbXKKyNe8RIn0vvepIHs= +github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc= +github.com/nats-io/nats.go v1.52.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= @@ -274,14 +290,15 @@ golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= -golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= -golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= diff --git a/internal/wsgateway/bridge.go b/internal/wsgateway/bridge.go index e1d007f..d9c4d70 100644 --- a/internal/wsgateway/bridge.go +++ b/internal/wsgateway/bridge.go @@ -26,12 +26,23 @@ func NewBridge(hub *Hub, sub events.Subscriber, log *zap.Logger) *Bridge { } // Run consumes events until ctx is cancelled. +// +// If the transport supports acknowledgements (NATS JetStream → AckSubscriber), +// the bridge uses the handler path so each event is ACK'd only after dispatch +// succeeds — genuine at-least-once. Otherwise it falls back to the channel path +// (Redis Pub/Sub → at-most-once). Either way dispatch logic is identical, so +// swapping EVENT_BUS needs no change here. func (b *Bridge) Run(ctx context.Context) error { + if ack, ok := b.sub.(events.AckSubscriber); ok { + b.log.Info("event bridge started", zap.String("mode", "ack")) + return ack.SubscribeAck(ctx, b.dispatch, events.TopicMatchmaking, events.TopicLeaderboard) + } + ch, err := b.sub.Subscribe(ctx, events.TopicMatchmaking, events.TopicLeaderboard) if err != nil { return err } - b.log.Info("event bridge started") + b.log.Info("event bridge started", zap.String("mode", "channel")) for { select { case <-ctx.Done(): @@ -40,16 +51,20 @@ func (b *Bridge) Run(ctx context.Context) error { if !ok { return nil } - b.dispatch(ctx, e) + _ = b.dispatch(ctx, e) } } } -func (b *Bridge) dispatch(ctx context.Context, e events.Event) { +// dispatch fans one event out to WebSocket clients. It returns an error only +// for transient/retryable failures so an AckSubscriber redelivers; poison +// inputs (malformed payloads) are dropped with a nil error so they are ACK'd +// and not redelivered forever. +func (b *Bridge) dispatch(ctx context.Context, e events.Event) error { // Extract the trace context the publisher embedded in the event and start a // consumer span linked to the producing trace, so matchmaking-tick -> - // publish -> ws-dispatch shows as one trace across Redis Pub/Sub. The hub - // fan-out below is ctx-free (in-memory), so we only need the span here. + // publish -> ws-dispatch shows as one trace across the async boundary. The + // hub fan-out below is ctx-free (in-memory), so we only need the span here. _, span := events.ReceiveSpan(ctx, e, "ws.dispatch") defer span.End() @@ -59,7 +74,7 @@ func (b *Bridge) dispatch(ctx context.Context, e events.Event) { if err := json.Unmarshal(e.Payload, &p); err != nil { tracing.RecordError(span, err) b.log.Warn("malformed MatchFound payload", zap.Error(err)) - return + return nil // poison message — drop, do not redeliver } span.SetAttributes(attribute.Int("ws.recipients", len(p.Players))) msg := marshal(Message{Type: events.TypeMatchFound, Room: p.RoomID, Data: e.Payload}) @@ -73,4 +88,5 @@ func (b *Bridge) dispatch(ctx context.Context, e events.Event) { default: b.log.Debug("ignoring event", zap.String("type", e.Type)) } + return nil } diff --git a/pkg/config/config.go b/pkg/config/config.go index 0632b98..f57bd0a 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -24,6 +24,13 @@ type Config struct { RedisPassword string RedisDB int + // Event transport selection. EventBus picks the messaging implementation + // ("redis" or "nats") at startup; services keep depending only on the + // events.Publisher/Subscriber interfaces, never on the concrete transport. + EventBus string + NATSURL string + EventWorkers int + JWTSecret string JWTExpiry time.Duration JWTIssuer string @@ -69,6 +76,12 @@ func Load(serviceName string) *Config { RedisPassword: getEnv("REDIS_PASSWORD", ""), RedisDB: getEnvInt("REDIS_DB", 0), + // Default to redis so a bare `go run` of a single service keeps working + // without NATS; docker-compose sets EVENT_BUS=nats explicitly. + EventBus: getEnv("EVENT_BUS", "redis"), + NATSURL: getEnv("NATS_URL", "nats://localhost:4222"), + EventWorkers: getEnvInt("EVENT_WORKERS", 8), + JWTSecret: getEnv("JWT_SECRET", "insecure-dev-secret-do-not-use-in-prod"), JWTExpiry: getEnvDuration("JWT_EXPIRY", 24*time.Hour), JWTIssuer: getEnv("JWT_ISSUER", "gamemesh"), diff --git a/pkg/events/events.go b/pkg/events/events.go index dc34c5a..ef2276c 100644 --- a/pkg/events/events.go +++ b/pkg/events/events.go @@ -76,7 +76,31 @@ type Publisher interface { } // Subscriber delivers events from one or more topics on a channel. +// +// This is a fire-and-forget push model with no acknowledgement hook: it fits +// at-most-once transports (Redis Pub/Sub) where a missed message is acceptable. +// Transports that support acknowledgements (NATS JetStream) ALSO implement +// AckSubscriber below; callers needing at-least-once type-assert for it. type Subscriber interface { Subscribe(ctx context.Context, topics ...string) (<-chan Event, error) Close() error } + +// Handler processes a single consumed event. Returning nil signals successful +// processing (the transport may ACK); a non-nil error signals failure (the +// transport may NAK and redeliver). Handlers must be safe for concurrent use: +// AckSubscriber implementations may invoke them from a bounded worker pool. +type Handler func(ctx context.Context, e Event) error + +// AckSubscriber is the richer, opt-in consumer contract for durable transports. +// It is intentionally separate from Subscriber so the existing interface — and +// every service depending on it — stays unchanged (additive evolution, not a +// breaking change). A transport such as NATSBus implements BOTH: Subscribe for +// drop-in compatibility, SubscribeAck for genuine at-least-once delivery. +// +// SubscribeAck delivers each event to handler and ACKs only after handler +// returns nil, giving "ACK after successful processing" semantics with retry on +// error. It blocks until ctx is cancelled, draining in-flight work on shutdown. +type AckSubscriber interface { + SubscribeAck(ctx context.Context, handler Handler, topics ...string) error +} diff --git a/pkg/events/events_test.go b/pkg/events/events_test.go index 80c5539..7605e45 100644 --- a/pkg/events/events_test.go +++ b/pkg/events/events_test.go @@ -35,7 +35,7 @@ func TestRedisBusRoundTrip(t *testing.T) { rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) t.Cleanup(func() { _ = rdb.Close() }) - bus := NewRedisBus(rdb, zap.NewNop()) + bus := NewRedisBus(rdb, zap.NewNop(), nil) t.Cleanup(func() { _ = bus.Close() }) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -62,7 +62,7 @@ func TestRedisBusDropsMalformedMessages(t *testing.T) { rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) t.Cleanup(func() { _ = rdb.Close() }) - bus := NewRedisBus(rdb, zap.NewNop()) + bus := NewRedisBus(rdb, zap.NewNop(), nil) t.Cleanup(func() { _ = bus.Close() }) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) diff --git a/pkg/events/factory.go b/pkg/events/factory.go new file mode 100644 index 0000000..3cd7822 --- /dev/null +++ b/pkg/events/factory.go @@ -0,0 +1,54 @@ +package events + +import ( + "fmt" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + + "github.com/alpnuhoglu/gamemesh/pkg/metrics" +) + +// Bus is the union of the transport capabilities. Every concrete transport +// (RedisBus, NATSBus) satisfies Publisher + Subscriber; NATSBus additionally +// satisfies AckSubscriber. Returning the union lets callers type-assert for the +// richer contract (the WS bridge does this) while still depending only on the +// narrow interfaces for the common path. +type Bus interface { + Publisher + Subscriber +} + +// Config selects and configures the event transport. It deliberately carries +// the few transport-agnostic knobs the factory needs; services never see it — +// they receive a Publisher/Subscriber. +type Config struct { + // Transport is "redis" or "nats". Unknown values are an error so a typo in + // EVENT_BUS fails loudly at startup rather than silently picking a default. + Transport string + + // DurableName identifies the consuming service for NATS durable consumers + // (ignored by Redis). E.g. "ws-bridge". + DurableName string + + // Workers bounds NATS consume-side concurrency (ignored by Redis). + Workers int +} + +// NewBus constructs the configured transport. It is the single place the +// transport switch lives, so no cmd/* entrypoint duplicates it and adding a +// future transport (Kafka, gRPC) touches one file. rdb is reused for the Redis +// transport; natsURL is dialed for the NATS transport. +// +// This is the seam that makes the migration a configuration change, not a code +// change: business services keep injecting the returned Publisher/Subscriber. +func NewBus(cfg Config, rdb *redis.Client, natsURL string, m *metrics.Metrics, log *zap.Logger) (Bus, error) { + switch cfg.Transport { + case "nats": + return NewNATSBus(natsURL, cfg.DurableName, cfg.Workers, m, log) + case "redis", "": + return NewRedisBus(rdb, log, m), nil + default: + return nil, fmt.Errorf("unknown EVENT_BUS %q (want \"redis\" or \"nats\")", cfg.Transport) + } +} diff --git a/pkg/events/nats.go b/pkg/events/nats.go new file mode 100644 index 0000000..3d3c000 --- /dev/null +++ b/pkg/events/nats.go @@ -0,0 +1,382 @@ +package events + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/propagation" + semconv "go.opentelemetry.io/otel/semconv/v1.26.0" + "go.opentelemetry.io/otel/trace" + "go.uber.org/zap" + + "github.com/alpnuhoglu/gamemesh/pkg/metrics" + "github.com/alpnuhoglu/gamemesh/pkg/tracing" +) + +// NATSBus implements Publisher, Subscriber and AckSubscriber over NATS +// JetStream. +// +// Why JetStream over Core NATS (and over the Redis Pub/Sub it replaces): +// - Durability: events are persisted to a stream, so a consumer that is down +// when an event is published still receives it on restart. Redis Pub/Sub is +// fire-and-forget — a missed message is gone. +// - At-least-once delivery: durable consumers + explicit ACKs guarantee an +// event is redelivered until processing succeeds. +// - Replay: a new or reset consumer can re-read the whole stream history +// (DeliverAll), which is impossible with Pub/Sub. +// - Outbox-ready: a future Outbox relay reuses Publish unchanged; the durable +// stream is the reliable hand-off point between DB and consumers. +// +// Redis stays the system of record for sessions, the matchmaking queue, the +// leaderboard sorted set and the room cache — JetStream only replaces the +// transport for inter-service events. +type NATSBus struct { + conn *nats.Conn + js jetstream.JetStream + log *zap.Logger + m *metrics.Metrics + workers int + durable string // durable consumer name prefix (per consuming service) + + mu sync.Mutex + cancels []context.CancelFunc + consumes []jetstream.ConsumeContext +} + +// stream describes the JetStream stream backing a topic family. Streams are +// per-domain (not one catch-all) so each domain's retention, limits and replay +// can be tuned independently and one domain's load never affects another. +type stream struct { + name string + subjects []string + maxAge time.Duration +} + +// streamFor maps a publish topic to the stream that should own it. Topics keep +// their existing values (events.matchmaking / events.leaderboard) so the event +// contract is unchanged; the stream subject is the hierarchical wildcard form +// (events.matchmaking.>) so future event subtypes fit without new streams. +var streamFor = map[string]stream{ + TopicMatchmaking: { + name: "MATCHMAKING", + subjects: []string{TopicMatchmaking, TopicMatchmaking + ".>"}, + maxAge: 24 * time.Hour, // match events are valuable to replay for a day + }, + TopicLeaderboard: { + name: "LEADERBOARD", + subjects: []string{TopicLeaderboard, TopicLeaderboard + ".>"}, + maxAge: 1 * time.Hour, // score updates are high-volume and disposable + }, +} + +// NewNATSBus connects to NATS, ensures the streams exist and returns a bus. +// durableName identifies the consuming service (e.g. "ws-bridge") so durable +// consumers survive restarts and resume from the last ACK'd message. workers +// bounds the consume-side concurrency. m may be nil (metrics skipped). +func NewNATSBus(url, durableName string, workers int, m *metrics.Metrics, log *zap.Logger) (*NATSBus, error) { + if workers <= 0 { + workers = 8 + } + conn, err := nats.Connect(url, + nats.Name("gamemesh-"+durableName), + nats.MaxReconnects(-1), // reconnect forever; JetStream tolerates gaps + nats.ReconnectWait(time.Second), + ) + if err != nil { + return nil, fmt.Errorf("nats connect: %w", err) + } + js, err := jetstream.New(conn) + if err != nil { + conn.Close() + return nil, fmt.Errorf("jetstream init: %w", err) + } + + b := &NATSBus{conn: conn, js: js, log: log, m: m, workers: workers, durable: durableName} + + // Ensure every known stream exists. CreateOrUpdateStream is idempotent, so + // every service can call this safely on boot regardless of start order. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + for _, s := range dedupeStreams() { + if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{ + Name: s.name, + Subjects: s.subjects, + Retention: jetstream.LimitsPolicy, + Storage: jetstream.FileStorage, + MaxAge: s.maxAge, + }); err != nil { + conn.Close() + return nil, fmt.Errorf("ensure stream %s: %w", s.name, err) + } + } + return b, nil +} + +// dedupeStreams returns the unique set of streams (streamFor maps multiple +// topics, but here each topic maps to a distinct stream; dedupe by name guards +// against future many-to-one mappings). +func dedupeStreams() []stream { + seen := map[string]bool{} + var out []stream + for _, s := range streamFor { + if !seen[s.name] { + seen[s.name] = true + out = append(out, s) + } + } + return out +} + +// Publish persists the event to its stream. It mirrors RedisBus.Publish: a +// producer span is started and the trace context is injected into the event's +// Carrier so the trace continues at the consumer — the mechanism is identical +// and transport-agnostic, only messaging.system differs. +func (b *NATSBus) Publish(ctx context.Context, topic string, e Event) error { + ctx, span := tracing.Tracer().Start(ctx, "events.publish "+topic, + trace.WithSpanKind(trace.SpanKindProducer), + trace.WithAttributes( + semconv.MessagingSystemKey.String("nats"), + semconv.MessagingDestinationName(topic), + attribute.String("messaging.event.type", e.Type), + ), + ) + defer span.End() + + if e.Carrier == nil { + e.Carrier = make(map[string]string) + } + otel.GetTextMapPropagator().Inject(ctx, propagation.MapCarrier(e.Carrier)) + + raw, err := json.Marshal(e) + if err != nil { + tracing.RecordError(span, err) + return err + } + + // Subject = topic. so it falls under the stream's wildcard and lets + // future per-type consumers filter (e.g. events.matchmaking.MatchFound) + // without changing the contract. + subject := topic + if e.Type != "" { + subject = topic + "." + e.Type + } + _, err = b.js.Publish(ctx, subject, raw) + tracing.RecordError(span, err) + if b.m != nil { + if err != nil { + b.m.EventsFailedTotal.WithLabelValues(topic, "publish").Inc() + } else { + b.m.EventsPublishedTotal.WithLabelValues(topic, e.Type).Inc() + } + } + return err +} + +// Subscribe satisfies the legacy channel-based Subscriber for drop-in +// compatibility. It internally uses SubscribeAck and ACKs as soon as the event +// is handed to the channel. Prefer SubscribeAck for true at-least-once: the +// channel model cannot signal processing success back to the transport. +func (b *NATSBus) Subscribe(ctx context.Context, topics ...string) (<-chan Event, error) { + out := make(chan Event, 256) + handler := func(_ context.Context, e Event) error { + select { + case out <- e: + return nil // ACK on enqueue — best-effort, like Pub/Sub + case <-ctx.Done(): + return ctx.Err() + } + } + if err := b.startConsumers(ctx, handler, topics...); err != nil { + return nil, err + } + // Close the channel when ctx ends so range-based consumers terminate. + go func() { + <-ctx.Done() + close(out) + }() + return out, nil +} + +// SubscribeAck delivers each event to handler from a bounded worker pool and +// ACKs only after handler returns nil — genuine "ACK after successful +// processing". On handler error the message is NAK'd with a short delay for +// redelivery (bounded by the consumer's MaxDeliver). It blocks until ctx is +// cancelled, then drains in-flight work and stops the consumers. +func (b *NATSBus) SubscribeAck(ctx context.Context, handler Handler, topics ...string) error { + if err := b.startConsumers(ctx, handler, topics...); err != nil { + return err + } + b.log.Info("jetstream consumers started", + zap.Strings("topics", topics), + zap.Int("workers", b.workers), + zap.String("durable", b.durable), + ) + <-ctx.Done() + return nil +} + +// startConsumers creates (or reuses) a durable pull consumer per topic and +// wires a bounded worker pool to process messages. Concurrency model: instead +// of one goroutine per message (unbounded, risks OOM under burst), we run a +// fixed pool of b.workers goroutines reading from an internal channel. +// jetstream.Consume already provides flow control / backpressure by capping the +// number of in-flight (unacked) messages to PullMaxMessages, so the pool plus +// AckWait bounds memory and prevents the consumer from outrunning processing. +func (b *NATSBus) startConsumers(ctx context.Context, handler Handler, topics ...string) error { + for _, topic := range topics { + s, ok := streamFor[topic] + if !ok { + return fmt.Errorf("no stream configured for topic %q", topic) + } + + // Durable name = service + stream, so each service has an independent + // cursor and restarts resume from the last ACK. + durable := sanitize(b.durable + "-" + strings.ToLower(s.name)) + cctx, err := b.js.CreateOrUpdateConsumer(ctx, s.name, jetstream.ConsumerConfig{ + Durable: durable, + AckPolicy: jetstream.AckExplicitPolicy, // manual ACK required + AckWait: 30 * time.Second, // redeliver if not ACK'd in time + MaxDeliver: 5, // bounded retries, then give up + DeliverPolicy: jetstream.DeliverAllPolicy, // replay history on first run + FilterSubject: topic + ".>", + }) + if err != nil { + return fmt.Errorf("ensure consumer %s: %w", durable, err) + } + + work := make(chan jetstream.Msg, b.workers) + var wg sync.WaitGroup + wg.Add(b.workers) + for i := 0; i < b.workers; i++ { + go func() { + defer wg.Done() + for msg := range work { + b.process(ctx, topic, handler, msg) + } + }() + } + + consume, err := cctx.Consume(func(msg jetstream.Msg) { + select { + case work <- msg: + case <-ctx.Done(): + _ = msg.Nak() // shutting down — let it redeliver later + } + }, jetstream.PullMaxMessages(b.workers*2)) // backpressure: cap in-flight + if err != nil { + close(work) + return fmt.Errorf("consume %s: %w", durable, err) + } + + // Graceful shutdown: when ctx ends, stop pulling, drain the pool, then + // stop the consume context. + shutdownCtx, cancel := context.WithCancel(ctx) + go func() { + <-shutdownCtx.Done() + consume.Stop() // stop delivering new messages + close(work) // let workers drain remaining buffered messages + wg.Wait() + }() + + b.mu.Lock() + b.cancels = append(b.cancels, cancel) + b.consumes = append(b.consumes, consume) + b.mu.Unlock() + } + return nil +} + +// process runs the handler for one message and translates its result into an +// ACK (success) or NAK-with-delay (failure → redelivery), recording metrics and +// a consumer span. This is the single point where at-least-once is enforced: +// the message is only removed from the stream's pending set after handler +// returns nil. +func (b *NATSBus) process(ctx context.Context, topic string, handler Handler, msg jetstream.Msg) { + var e Event + if err := json.Unmarshal(msg.Data(), &e); err != nil { + b.log.Warn("dropping malformed event", zap.Error(err), zap.String("subject", msg.Subject())) + if b.m != nil { + b.m.EventsFailedTotal.WithLabelValues(topic, "decode").Inc() + } + _ = msg.Term() // poison message — never redeliver + return + } + + // Continue the producer's trace across the async boundary (same mechanism + // as RedisBus / ReceiveSpan, just inlined so we can wrap ACK timing). + hctx := ctx + if e.Carrier != nil { + hctx = otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(e.Carrier)) + } + hctx, span := tracing.Tracer().Start(hctx, "events.consume "+topic, + trace.WithSpanKind(trace.SpanKindConsumer), + trace.WithAttributes( + semconv.MessagingSystemKey.String("nats"), + attribute.String("messaging.event.type", e.Type), + ), + ) + defer span.End() + + start := time.Now() + err := handler(hctx, e) + if b.m != nil { + b.m.EventProcessingDuration.WithLabelValues(topic, e.Type).Observe(time.Since(start).Seconds()) + } + + if err != nil { + tracing.RecordError(span, err) + b.log.Warn("event handler failed, will redeliver", + zap.Error(err), zap.String("type", e.Type), zap.String("id", e.ID)) + if b.m != nil { + b.m.EventsFailedTotal.WithLabelValues(topic, "handle").Inc() + } + // NAK with backoff; JetStream redelivers up to MaxDeliver. + _ = msg.NakWithDelay(2 * time.Second) + return + } + + if ackErr := msg.Ack(); ackErr != nil { + // ACK failed (e.g. connection blip). The message will redeliver on + // AckWait expiry — at-least-once holds, handler must be idempotent. + b.log.Warn("ack failed; message will redeliver", zap.Error(ackErr), zap.String("id", e.ID)) + tracing.RecordError(span, ackErr) + return + } + if b.m != nil { + b.m.EventsConsumedTotal.WithLabelValues(topic, e.Type).Inc() + } +} + +// Close stops all consumers and drains the NATS connection. +func (b *NATSBus) Close() error { + b.mu.Lock() + for _, cancel := range b.cancels { + cancel() + } + for _, c := range b.consumes { + c.Stop() + } + b.cancels = nil + b.consumes = nil + b.mu.Unlock() + + // Drain flushes pending publishes and waits for in-flight callbacks. + if err := b.conn.Drain(); err != nil && !errors.Is(err, nats.ErrConnectionClosed) { + return err + } + return nil +} + +// sanitize makes a string safe for use as a NATS durable name (no dots/spaces). +func sanitize(s string) string { + return strings.NewReplacer(".", "_", " ", "_", "*", "_", ">", "_").Replace(s) +} diff --git a/pkg/events/nats_test.go b/pkg/events/nats_test.go new file mode 100644 index 0000000..0b187cf --- /dev/null +++ b/pkg/events/nats_test.go @@ -0,0 +1,146 @@ +package events + +import ( + "context" + "sync/atomic" + "testing" + "time" + + natsserver "github.com/nats-io/nats-server/v2/server" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/trace" + "go.uber.org/zap" +) + +// startEmbeddedJetStream boots an in-process NATS server with JetStream enabled +// so these tests need no docker. It returns the client URL. +func startEmbeddedJetStream(t *testing.T) string { + t.Helper() + opts := &natsserver.Options{ + Host: "127.0.0.1", + Port: -1, // random free port + JetStream: true, + StoreDir: t.TempDir(), + } + srv, err := natsserver.NewServer(opts) + require.NoError(t, err) + go srv.Start() + if !srv.ReadyForConnections(5 * time.Second) { + t.Fatal("embedded NATS server not ready") + } + t.Cleanup(srv.Shutdown) + return srv.ClientURL() +} + +// TestNATSPublishConsumeAck is the happy path: an event published to JetStream +// is delivered to the handler, processed, and acknowledged exactly with the +// trace context preserved across the async boundary. +func TestNATSPublishConsumeAck(t *testing.T) { + setupTracing(t) + url := startEmbeddedJetStream(t) + + bus, err := NewNATSBus(url, "test", 4, nil, zap.NewNop()) + require.NoError(t, err) + t.Cleanup(func() { _ = bus.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + gotTrace := make(chan trace.TraceID, 1) + handler := func(hctx context.Context, e Event) error { + gotTrace <- trace.SpanContextFromContext(hctx).TraceID() + return nil + } + go func() { _ = bus.SubscribeAck(ctx, handler, TopicMatchmaking) }() + // Give the consumer a moment to attach. + time.Sleep(300 * time.Millisecond) + + // Publish inside an active parent span so we can assert trace continuity. + pctx, parent := otel.Tracer("test").Start(ctx, "parent") + want := parent.SpanContext().TraceID() + e, err := New(TypeMatchFound, MatchFoundPayload{RoomID: "r1", Players: []string{"p1", "p2"}}) + require.NoError(t, err) + require.NoError(t, bus.Publish(pctx, TopicMatchmaking, e)) + parent.End() + + select { + case got := <-gotTrace: + assert.Equal(t, want, got, "consumer handler must run under the producer's trace") + case <-ctx.Done(): + t.Fatal("timed out waiting for event to be consumed") + } +} + +// TestNATSRedeliversOnHandlerError asserts at-least-once: a handler that fails +// the first time causes a NAK and the event is redelivered until it succeeds. +func TestNATSRedeliversOnHandlerError(t *testing.T) { + setupTracing(t) + url := startEmbeddedJetStream(t) + + bus, err := NewNATSBus(url, "retry", 1, nil, zap.NewNop()) + require.NoError(t, err) + t.Cleanup(func() { _ = bus.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + var attempts int32 + done := make(chan struct{}) + handler := func(_ context.Context, _ Event) error { + if atomic.AddInt32(&attempts, 1) < 2 { + return assert.AnError // fail first delivery -> NAK -> redeliver + } + close(done) + return nil + } + go func() { _ = bus.SubscribeAck(ctx, handler, TopicLeaderboard) }() + time.Sleep(300 * time.Millisecond) + + e, err := New(TypeLeaderboardUpdated, LeaderboardUpdatedPayload{PlayerID: "p1", Score: 9, Rank: 1}) + require.NoError(t, err) + require.NoError(t, bus.Publish(ctx, TopicLeaderboard, e)) + + select { + case <-done: + assert.GreaterOrEqual(t, atomic.LoadInt32(&attempts), int32(2), "event must be redelivered after a failed handler") + case <-ctx.Done(): + t.Fatal("timed out waiting for redelivery") + } +} + +// TestNATSReplay asserts replay: an event published before any consumer exists +// is still delivered, because JetStream persists it (DeliverAll). This is the +// capability Redis Pub/Sub lacks. +func TestNATSReplay(t *testing.T) { + setupTracing(t) + url := startEmbeddedJetStream(t) + + bus, err := NewNATSBus(url, "replay", 1, nil, zap.NewNop()) + require.NoError(t, err) + t.Cleanup(func() { _ = bus.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // Publish first, with NO consumer attached yet. + e, err := New(TypeMatchFound, MatchFoundPayload{RoomID: "r9", Players: []string{"p1"}}) + require.NoError(t, err) + require.NoError(t, bus.Publish(ctx, TopicMatchmaking, e)) + + // Now attach a fresh consumer; it must still receive the earlier event. + got := make(chan string, 1) + handler := func(_ context.Context, ev Event) error { + got <- ev.ID + return nil + } + go func() { _ = bus.SubscribeAck(ctx, handler, TopicMatchmaking) }() + + select { + case id := <-got: + assert.Equal(t, e.ID, id, "consumer must replay an event published before it existed") + case <-ctx.Done(): + t.Fatal("timed out waiting for replayed event") + } +} diff --git a/pkg/events/redis.go b/pkg/events/redis.go index 83d4a31..aaf5492 100644 --- a/pkg/events/redis.go +++ b/pkg/events/redis.go @@ -12,6 +12,7 @@ import ( "go.opentelemetry.io/otel/trace" "go.uber.org/zap" + "github.com/alpnuhoglu/gamemesh/pkg/metrics" "github.com/alpnuhoglu/gamemesh/pkg/tracing" ) @@ -25,12 +26,14 @@ import ( type RedisBus struct { client *redis.Client log *zap.Logger + m *metrics.Metrics subs []*redis.PubSub } -// NewRedisBus wraps an existing Redis client. -func NewRedisBus(client *redis.Client, log *zap.Logger) *RedisBus { - return &RedisBus{client: client, log: log} +// NewRedisBus wraps an existing Redis client. m may be nil (e.g. in unit tests), +// in which case metrics recording is skipped — the bus stays usable. +func NewRedisBus(client *redis.Client, log *zap.Logger, m *metrics.Metrics) *RedisBus { + return &RedisBus{client: client, log: log, m: m} } // Publish marshals and publishes the event to a topic. It creates a producer @@ -60,6 +63,13 @@ func (b *RedisBus) Publish(ctx context.Context, topic string, e Event) error { } err = b.client.Publish(ctx, topic, raw).Err() tracing.RecordError(span, err) + if b.m != nil { + if err != nil { + b.m.EventsFailedTotal.WithLabelValues(topic, "publish").Inc() + } else { + b.m.EventsPublishedTotal.WithLabelValues(topic, e.Type).Inc() + } + } return err } @@ -99,8 +109,14 @@ func (b *RedisBus) Subscribe(ctx context.Context, topics ...string) (<-chan Even var e Event if err := json.Unmarshal([]byte(msg.Payload), &e); err != nil { b.log.Warn("dropping malformed event", zap.Error(err), zap.String("topic", msg.Channel)) + if b.m != nil { + b.m.EventsFailedTotal.WithLabelValues(msg.Channel, "decode").Inc() + } continue } + if b.m != nil { + b.m.EventsConsumedTotal.WithLabelValues(msg.Channel, e.Type).Inc() + } select { case out <- e: case <-ctx.Done(): diff --git a/pkg/events/trace_propagation_test.go b/pkg/events/trace_propagation_test.go index 903006c..afc9a11 100644 --- a/pkg/events/trace_propagation_test.go +++ b/pkg/events/trace_propagation_test.go @@ -43,7 +43,7 @@ func TestPublishInjectsTraceContext(t *testing.T) { mr := miniredis.RunT(t) rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) t.Cleanup(func() { _ = rdb.Close() }) - bus := NewRedisBus(rdb, zap.NewNop()) + bus := NewRedisBus(rdb, zap.NewNop(), nil) t.Cleanup(func() { _ = bus.Close() }) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 0f97429..d462cd2 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -25,6 +25,15 @@ type Metrics struct { MatchmakingQueueSize prometheus.Gauge MatchesCreated prometheus.Counter LeaderboardUpdates prometheus.Counter + + // Event-transport instruments. Shared by every transport (RedisBus, + // NATSBus) so dashboards work regardless of EVENT_BUS. Labelled by topic so + // matchmaking and leaderboard traffic can be split; result distinguishes + // ack/nak/error on the consume path. + EventsPublishedTotal *prometheus.CounterVec + EventsConsumedTotal *prometheus.CounterVec + EventsFailedTotal *prometheus.CounterVec + EventProcessingDuration *prometheus.HistogramVec } // New creates and registers all instruments, labelled with the service name @@ -76,11 +85,33 @@ func New(service string) *Metrics { Help: "Total leaderboard score updates.", ConstLabels: constLabels, }), + EventsPublishedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gamemesh_events_published_total", + Help: "Total events published to the event bus.", + ConstLabels: constLabels, + }, []string{"topic", "type"}), + EventsConsumedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gamemesh_events_consumed_total", + Help: "Total events successfully consumed and acknowledged.", + ConstLabels: constLabels, + }, []string{"topic", "type"}), + EventsFailedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gamemesh_events_failed_total", + Help: "Total event publish/consume failures.", + ConstLabels: constLabels, + }, []string{"topic", "stage"}), + EventProcessingDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "gamemesh_event_processing_duration_seconds", + Help: "Handler processing latency for consumed events.", + ConstLabels: constLabels, + Buckets: []float64{.001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5}, + }, []string{"topic", "type"}), } reg.MustRegister( m.RequestsTotal, m.ErrorsTotal, m.RequestDuration, m.WSConnections, m.MatchmakingQueueSize, m.MatchesCreated, m.LeaderboardUpdates, + m.EventsPublishedTotal, m.EventsConsumedTotal, m.EventsFailedTotal, m.EventProcessingDuration, ) return m } diff --git a/tests/integration/redis_test.go b/tests/integration/redis_test.go index 552dcc6..c682350 100644 --- a/tests/integration/redis_test.go +++ b/tests/integration/redis_test.go @@ -81,7 +81,7 @@ func TestMatchmakingQueueAgainstRealRedis(t *testing.T) { func TestRedisBusPubSub(t *testing.T) { rdb := startRedis(t) - bus := events.NewRedisBus(rdb, zap.NewNop()) + bus := events.NewRedisBus(rdb, zap.NewNop(), nil) t.Cleanup(func() { _ = bus.Close() }) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)