NATS and JetStream

FluxaORM uses NATS JetStream as its messaging transport. Three features are built on top of it:

• Entity change events - every write to an entity tagged orm:"cdc" is published as a message.
• Consumers - typed, generated handlers that read entity events and tasks.
• Tasks - background jobs dispatched onto queues, with a run history table.

A NATS pool is a connection to one or more NATS servers. This page covers the pool itself, the low-level publish and fetch API, hand-registered streams and consumers, and the topology reconciliation (GetNatsAlters) that creates and updates streams and durable consumers on the broker.

Pool Registration

Register a pool with RegisterNats:

import "github.com/latolukasz/fluxaorm/v2"

registry := fluxaorm.NewRegistry()

// Pool with default options (nil is allowed and becomes an empty NatsPoolOptions):
registry.RegisterNats([]string{"nats://localhost:4222"}, fluxaorm.DefaultPoolCode, nil)

// Pool with several servers and options:
registry.RegisterNats([]string{"nats://nats-1:4222", "nats://nats-2:4222"}, "events", &fluxaorm.NatsPoolOptions{
    ClientID:             "my-service",
    RetryOnFailedConnect: true,
    Auth: &fluxaorm.NatsAuthConfig{
        User:     "app",
        Password: "secret",
    },
})

RegisterNats(urls []string, poolCode string, options *NatsPoolOptions)

The pool is available from the engine as engine.Nats(code) (returns nil for an unknown code) and all pools as engine.Registry().NatsPools().

The features built on NATS pick their pool with the NatsPool field of EntityStreamOptions, TaskStreamOptions and ConsumerDef. All three default to fluxaorm.DefaultPoolCode ("default"), so registering the pool under the default code needs no further wiring.

NatsPoolOptions

NatsAuthConfig

Exactly one method is used, in this order of precedence:

YAML Configuration

Pools can also come from a configuration file. The list-based shape below is the fluxaorm.Config struct (yaml tags) consumed by registry.InitByConfig - see Registry. The top-level key is natsPools:

natsPools:
  - code: default
    urls: ["nats://localhost:4222"]
    clientID: my-service
    maxReconnects: -1
    reconnectWaitMs: 1000
    reconnectBufSize: 8388608
    connectTimeoutMs: 5000
    retryOnFailedConnect: true
    authUser: app
    authPassword: secret
    ignoredSubjects: ["legacy.>"]
    streams:
      - name: ORDERS_RAW
        subjects: ["orders.>"]
        maxAgeMs: 604800000
        maxBytes: 1073741824
        maxMsgSize: 1048576
        replicas: 1
        duplicateWindowMs: 600000
    consumers:
      - name: orders-audit
        filterSubjects: ["orders.>"]
        ackWaitMs: 30000
        maxAckPending: 16
        maxDeliver: -1

Per-pool keys: code (required), urls (required), clientID, maxReconnects, reconnectWaitMs, reconnectBufSize, connectTimeoutMs, retryOnFailedConnect, authToken, authUser, authPassword, authCredsFile, authNKeySeed, ignoredSubjects, ignoredConsumers, streams, consumers.

streams entries: name (required), subjects (required), maxAgeMs, maxBytes, maxMsgSize, replicas, duplicateWindowMs, storage, retention. Each becomes a NewNatsStream(name, code) builder registered with RegisterNatsStream.

consumers entries: name (required), filterSubjects, ackWaitMs, maxAckPending, maxDeliver. Each becomes a NewNatsConsumer(name, code) builder registered with RegisterNatsConsumer.

Publishing

The Nats interface is what engine.Nats(code) returns:

type Nats interface {
    GetCode() string
    GetURLs() []string
    GetPoolOptions() *NatsPoolOptions
    Ping() error
    Publish(ctx Context, msg *NatsMessage) error
    PublishWithAck(ctx Context, msg *NatsMessage) (NatsPubAck, error)
    PublishBatch(ctx Context, msgs []*NatsMessage) error
    PublishAsync(ctx Context, msg *NatsMessage, callback func(*NatsMessage, error))
    Consumer(name string) (NatsConsumer, error)
    MustConsumer(name string) NatsConsumer
    ConsumerNames() []string
    GetJetStream() (jetstream.JetStream, error)
    GetConn() (*nats.Conn, error)
    Close()
}

Messages are NatsMessage values:

type NatsMessage struct {
    Subject    string
    Data       []byte
    Headers    nats.Header
    Sequence   uint64    // stream sequence, set on fetched messages
    Timestamp  time.Time // broker timestamp, set on fetched messages
    Deliveries uint64    // how many times JetStream delivered it (starts at 1); 0 on a message you build
}

func NewNatsMessage(subject string) *NatsMessage // Headers is pre-initialised

ctx := engine.NewContext(context.Background())
pool := engine.Nats("events")

msg := fluxaorm.NewNatsMessage("orders.created")
msg.Data = []byte(`{"id":1}`)
msg.Headers.Set("Nats-Msg-Id", "order-created-1") // optional JetStream deduplication

ack, err := pool.PublishWithAck(ctx, msg)
if err != nil {
    panic(err)
}
if ack.Duplicate {
    fmt.Println("collapsed by the stream's duplicate window")
}

Streams and Consumers

FluxaORM manages two streams of its own (FLUXA_ENTITY and FLUXA_TASK, see below). Any other stream or durable consumer you need is declared with the builders and reconciled onto the broker by GetNatsAlters.

NatsStreamBuilder

registry.RegisterNatsStream(
    fluxaorm.NewNatsStream("ORDERS_RAW", "events").
        Subjects("orders.>").
        MaxAge(7 * 24 * time.Hour).
        Duplicates(10 * time.Minute),
)

Validate() fails with nats stream name is required, nats pool code is required for stream '%s', nats stream '%s' must have at least one subject, nats pool '%s' not registered for stream '%s' or duplicate nats stream '%s' in pool '%s' (already registered in pool '%s').

NatsConsumerBuilder

registry.RegisterNatsConsumer(
    fluxaorm.NewNatsConsumer("orders-audit", "events").
        Stream("ORDERS_RAW").
        FilterSubjects("orders.>").
        AckWait(time.Minute).
        MaxAckPending(16),
)

The ack policy is always AckExplicitPolicy. Validate() fails with nats consumer name is required, nats pool code is required for consumer '%s', nats pool '%s' not registered for consumer '%s' or duplicate nats consumer '%s' in pool '%s' (already registered in pool '%s').

Fetching messages

A registered consumer is obtained from the pool by name and drained with Fetch:

type NatsConsumer interface {
    GetName() string
    GetSettings() *NatsConsumerSettings
    Fetch(ctx Context, batch int, maxWait time.Duration) NatsBatch
    Close()
}

consumer, err := engine.Nats("events").Consumer("orders-audit")
if err != nil {
    panic(err) // "nats pool 'events': consumer 'orders-audit' not registered"
}

for {
    batch := consumer.Fetch(ctx, 50, 2*time.Second)
    if err := batch.Error(); err != nil {
        log.Printf("fetch: %v", err)
        continue
    }
    for _, msg := range batch.Records() {
        if err := handle(msg); err != nil {
            _ = msg.NakWithDelay(10 * time.Second) // redeliver later
            continue
        }
        _ = msg.Ack()
    }
}

• Consumer(name) knows the durables registered with RegisterNatsConsumer and the ones declared through ConsumerDef. MustConsumer panics instead of returning the error; ConsumerNames() lists them (unsorted).
• GetSettings() returns the registered definition: Name, FilterSubjects []string, AckWait, MaxAckPending, MaxDeliver and DeliverPolicy (a ConsumerDef durable leaves DeliverPolicy at its zero value, DeliverAllPolicy).
• On the first Fetch the durable is located by scanning every stream on the server; if none has it the batch error is nats pool '%s': durable consumer '%s' not found in any stream. Run GetNatsAlters first.
• A fetch that times out with no messages is an idle poll, not an error: batch.Error() is nil and batch.IsEmpty() is true.
• NatsBatch offers Records() []*NatsMessage, EachRecord(func(*NatsMessage)), EachError(func(error)), Error() error and IsEmpty() bool.
• Fetched messages carry Sequence, Timestamp and Deliveries from the JetStream metadata.
• msg.Ack(), msg.Nak(), msg.NakWithDelay(d) and msg.Term() act on the JetStream message. On a message that was not fetched (one you built yourself) they are no-ops returning nil.
• NatsConsumer.Close() is a no-op; close the pool instead.

Streams Managed by FluxaORM

When you use entity events or tasks, GetNatsAlters also manages two streams on the pool selected by EntityStreamOptions.NatsPool / TaskStreamOptions.NatsPool:

Both streams are tuned with EntityStreamOptions / TaskStreamOptions (fields Storage, Replicas, MaxAge, MaxBytes, DuplicateWindow, NatsPool; defaults: pool default, Replicas=1, MaxAge=7d, DuplicateWindow=10m, MaxBytes unbounded, Storage file). See Entity events and Tasks.

Subjects are typed as fluxaorm.Subject and built with:

Related string types: ConsumerName (also the JetStream durable name), Queue, TaskName, and the constant DefaultQueue Queue = "default".

Topology Alters

GetNatsAlters compares the desired topology with the broker and returns the operations needed to synchronise them. It is the NATS counterpart of GetAlters for MySQL - the two are separate, run both:

func GetNatsAlters(ctx Context) ([]NatsAlter, error)

type NatsAlter struct {
    Description string      // e.g. "CREATE nats stream 'FLUXA_ENTITY'"
    PoolCode    string
    Kind        AlterKind   // AlterKindCreateTable, AlterKindModifyTable or AlterKindDropTable
    Safety      AlterSafety // AlterSafe or AlterDestructive
}

func (a NatsAlter) IsSafe() bool
func (a NatsAlter) Exec(ctx Context) error

ctx := engine.NewContext(context.Background())

// MySQL schema first (tables), then the JetStream topology.
alters, err := fluxaorm.GetAlters(ctx)
if err != nil {
    panic(err)
}
for _, alter := range alters {
    if err := alter.Exec(ctx); err != nil {
        panic(err)
    }
}

natsAlters, err := fluxaorm.GetNatsAlters(ctx)
if err != nil {
    panic(err)
}
for _, alter := range natsAlters {
    if !alter.IsSafe() {
        // Stream deletions and subject-narrowing updates break the previous
        // code version if it is still running. Gate them behind a manual step.
        fmt.Printf("skipping destructive alter: %s (pool %s)\n", alter.Description, alter.PoolCode)
        continue
    }
    if err := alter.Exec(ctx); err != nil {
        panic(err)
    }
}

The desired state is: every RegisterNatsStream builder, plus FLUXA_ENTITY and FLUXA_TASK when applicable; every RegisterNatsConsumer builder, plus one durable per ConsumerDef (pinned to its stream, filtering all its subjects, with the ConsumerDef's AckWait, MaxAckPending, MaxDeliver).

Details worth knowing:

• Two stream configs are equal when Retention, Storage, Replicas, MaxAge, MaxBytes, MaxMsgSize, Duplicates and the (sorted) subject lists match.
• Streams whose names do not start with FLUXA_/fluxa_ are never deleted, whatever else is on the broker. A managed stream that carries a subject listed in the pool's IgnoredSubjects is not deleted either.
• Consumers are never diffed or deleted. ENSURE is always reported as safe and updates the durable in place, so widening a consumer's filter list (for example adding an entity to a ConsumerDef) keeps its position in the stream. A change that narrows the filter is applied too and will stop deliveries the previous code version still expects - it is not flagged.
• GetNatsAlters connects to every pool that has at least one desired stream. A pool that has desired consumers but no desired streams is skipped entirely.
• The result is sorted by Description, which places CREATE before ENSURE before UPDATE.

Logging and Metrics

NATS operations are logged when a query logger is registered with Nats: true (see Queries log):

ctx.RegisterQueryLogger(myHandler, fluxaorm.QueryLoggerOptions{Nats: true})
// or, to stderr:
ctx.EnableQueryDebugCustom(fluxaorm.QueryLoggerOptions{Nats: true})

Log source is nats. Operations: PUBLISH (query subject: <subject>), PUBLISH_BATCH (messages: <n>), PUBLISH_ASYNC, FETCH (<n> messages fetched, pool label <pool>/<consumer>), and from the consumer runtime CDC_HANDLE / TASK_HANDLE (handler errors, pool label is the consumer name, no duration).

Prometheus metrics (see Metrics):

Closing

engine.Nats(code).Close() closes the underlying connection. The next operation on the pool reconnects lazily, so call it on shutdown only:

defer engine.Nats(fluxaorm.DefaultPoolCode).Close()