nats

README

nats

NATS for the phpboyscout estate: a go/controls-managed embedded server, and a client that is identical whether the broker is in this process or a cluster somebody else runs.

go get gitlab.com/phpboyscout/go/nats

!!! danger "v0.1.0 runs in-process only"

An embedded server with a network listener is **refused**, including cluster routes. This module
has no TLS or authentication configuration yet, so a listening server would accept anything that
could reach it. Refusing means the unsafe surface does not exist rather than existing with a
warning beside it. Connecting a client to a cluster somebody else operates and secures is
unaffected, and is what most services should do.

The problem it solves

A service that needs messaging early needs it between two goroutines, and a service that needs it later needs it between two clusters. Those are usually two pieces of code, written a year apart, and the second is a rewrite.

NATS makes them the same code — an embedded server is the same server, configured differently — and this module makes the seam an estate convention rather than something each project rediscovers.

Wiring

Register the server first, then the client. go/controls starts its services concurrently, so the client may dial before the server has finished starting; an in-process dial waits for it, up to ServerSettings.ReadyTimeout. It stops them in reverse registration order, which is what the order is for: the client drains before the server goes away underneath it.

srv, err := server.Register(ctx, "nats", controller, log, nats.ServerSettings{
    InProcessOnly: true,
})

cli, err := client.Register(ctx, "nats-client", controller,
    nats.ClientSettings{Name: "my-service"}, nil,
    client.InProcess(srv))

The client starts trying when the controller starts it, not when it is built, and Start succeeds even if the broker is not there yet — WaitReady(ctx) is how you wait for it. Settings are validated at wiring, so a misconfiguration fails where an operator is watching; every method reports ErrNotConnected until the connection exists.

To reach a cluster instead, drop the server and set ClientSettings.URL. Nothing that publishes or subscribes changes.

The two packages

Package	What it is
nats	The settings both halves are built from, and nothing else
nats/server	An embedded server with a controls-compatible Register
nats/client	A connection, in-process or over the network
nats/natstest	A real server and client for a test

They are separate because most consumers want only the second, and importing nats/client alone does not pull a broker into the binary. A guard test asserts it.

What it adds over nats.Connect

Refusals rather than features.

Every subscription is bounded, and the bound cannot be disabled. The client library defaults to 500,000 pending messages, which is not a bound so much as a promise to run out of memory at a moment nobody chose. Non-positive limits are rejected.

Every asynchronous error is reported. NATS discards from a full subscription queue rather than delaying the publisher, and reports it out of band — as it also reports publish permission failures, which arrive after Publish has returned nil. The logger defaults to slog.Default(), because the consumer most likely to lose messages unnoticed is the one that configured nothing.

A connection has one target, and it cannot carry a secret. A URL with userinfo is refused, and so is a credential over an unencrypted scheme.

Close waits. nats.Conn.Drain() starts a goroutine and returns; a Close built on that alone reports success while handlers are still running.

Testing against it

cli := natstest.Start(t)

A real server and a connected client, cleaned up in the right order, with no broker to deploy and nothing to mock. See Test a service that uses NATS.

Configuration

This module does not read configuration. It takes typed settings, and the calling application maps its own configuration onto them — which is what keeps go/config out of every module's dependency graph, and lets one application configure from YAML while another configures from flags.

Documentation

https://nats.go.phpboyscout.uk

Licence

MIT. See LICENSE.

Documentation

Overview

Package nats carries the settings the estate's NATS server and client are constructed from, and nothing else.

Why the settings live apart from the code that uses them

Toolkit modules do not read configuration. They take typed settings, and the calling application maps its own configuration source onto them — that is what keeps go/config out of every module's dependency graph and lets one application configure a module from YAML while another configures it from flags. The struct tags exist so that mapping is a decode rather than a hand-written assignment.

The two halves

nats/server  an embedded NATS server, registered with go/controls
nats/client  a connection, in-process or over the network

They are separate packages because most consumers want only the second. A service that talks to a NATS cluster somebody else operates should not pull a broker into its binary to do it.

The progression this module exists to make boring

A single process with no socket at all, then a listener, then a mesh with peers, then an external cluster. In NATS these are the same server with different options, and the same client with a different address — so this module's job is to make sure the code above it never learns which one it is talking to.

Index

• Constants
• Variables
• func Reconnects(n int) *int
• type ClientSettings
  • func (s ClientSettings) WithDefaults() ClientSettings
• type CredentialSource
  • func StaticCredential(token string) CredentialSource
• type ServerSettings
  • func (s ServerSettings) WithDefaults() ServerSettings

Constants

const (
	// DefaultPendingMsgs bounds a subscription's queue in messages.
	//
	// Deliberately far below the NATS client default of 500,000. A limit that
	// large is not a limit — it is a promise to run out of memory later, at a
	// moment nobody chose. A consumer that genuinely needs a deeper queue can
	// say so; one that has never thought about it gets a number small enough
	// that shedding is visible while the traffic is still explicable.
	DefaultPendingMsgs = 4096

	// DefaultPendingBytes bounds the same queue in bytes, because message count
	// alone does not bound memory when payload size is not controlled.
	DefaultPendingBytes = 32 << 20 // 32 MiB

	// DefaultReconnectWait is how long the client waits between attempts.
	DefaultReconnectWait = 2 * time.Second

	// DefaultMaxReconnects is negative on purpose: retry forever. A long-running
	// service that stops trying to reconnect is a service that has to be
	// restarted by a human to recover from a broker outage it would have
	// survived.
	DefaultMaxReconnects = -1

	// DefaultDrainTimeout bounds a drain whose context carries no deadline.
	//
	// Close waits for the drain to finish, which is the whole reason it exists.
	// But a caller passing context.Background() would then wait for ever — the
	// unbounded hang moved rather than removed — so a deadline-free context
	// gets this instead. NATS' own drain timeout is 30 seconds; this is shorter
	// because a shutdown nobody bounded is one somebody is waiting on.
	DefaultDrainTimeout = 10 * time.Second

	// DefaultReadyTimeout bounds how long Start waits for an embedded server to
	// accept connections before reporting failure.
	DefaultReadyTimeout = 10 * time.Second
)

Defaults for the settings that have a safe one. A store directory and a credential deliberately do not.

Variables

var (
	// ErrNetworkModeUnsupported is returned when an embedded server is asked to
	// listen on a network interface or mesh with peers.
	//
	// This module has no TLS or authentication configuration yet, so a
	// listening server would accept any connection that could reach it, with no
	// credential and no encryption — and a token supplied by a client would not
	// be required by the server, nor protected in transit.
	//
	// Refusing is the honest v0.1.0 position: the unsafe surface does not exist
	// rather than existing with a warning next to it. In-process servers are
	// unaffected, and a client can still reach a cluster somebody else operates
	// and secures.
	ErrNetworkModeUnsupported = errors.NewSentinel("nats.network_mode_unsupported",
		"nats: an embedded server can only run in-process until TLS and authentication are supported")

	// ErrCredentialInURL is returned when a URL carries userinfo.
	//
	// NATS accepts credentials in a connection URL, so `nats://token@broker`
	// puts a secret in a settings struct that gets logged, dumped and committed.
	// The credential belongs in a CredentialSource, which is resolved at connect
	// time and never stored.
	ErrCredentialInURL = errors.NewSentinel("nats.credential_in_url",
		"nats: the URL carries a credential; use a CredentialSource instead")

	// ErrInsecureCredential is returned when a credential would be sent over an
	// unencrypted connection.
	//
	// A NATS token is bearer material. Over `nats://` it crosses the network in
	// plaintext, so anyone on the path has it. Either use `tls://`, or connect
	// without a credential to a server that does not require one.
	ErrInsecureCredential = errors.NewSentinel("nats.insecure_credential",
		"nats: refusing to send a credential over an unencrypted connection")

	// ErrUnboundedSubscription is returned when a subscription would be created
	// with no bound.
	//
	// The module's one non-negotiable guarantee. A non-positive limit means
	// "unlimited" to NATS, so accepting it would let a blocked handler grow a
	// queue until the process dies — which is the failure the bound exists to
	// convert into a visible shed.
	ErrUnboundedSubscription = errors.NewSentinel("nats.unbounded_subscription",
		"nats: pending message and byte limits must both be positive")

	// ErrInvalidSetting is returned when a setting cannot mean anything.
	//
	// Separate from the specific refusals because those are policy — this one is
	// arithmetic. A negative timeout or a negative limit is not a configuration
	// this module disagrees with; it is one that cannot be honoured, and saying
	// so at construction is the difference between a named mistake and a
	// runtime failure that looks like a broken broker.
	ErrInvalidSetting = errors.NewSentinel("nats.invalid_setting",
		"nats: setting cannot be negative")

	// ErrNotConnected is returned by a client that has not connected yet, or
	// whose connection has closed.
	//
	// A client registered with the controller connects when the controller
	// starts it, not when it is built — so "not yet" is an ordinary state
	// during wiring rather than a failure, and worth distinguishing from a
	// connection that was lost.
	ErrNotConnected = errors.NewSentinel("nats.not_connected",
		"nats: client is not connected")

	// ErrNoStoreDir is returned when JetStream is enabled with no store
	// directory. A JetStream server with no store directory writes wherever it
	// decides to, which differs by host and survives a deployment differently
	// on each one.
	ErrNoStoreDir = errors.NewSentinel("nats.no_store_dir",
		"nats: jetstream requires a store directory")

	// ErrNoClusterName is returned when routes are configured with no cluster
	// name. A server with no cluster name cannot mesh.
	ErrNoClusterName = errors.NewSentinel("nats.no_cluster_name",
		"nats: routes require a cluster name")

	// ErrNoTarget is returned when a client has neither a URL nor an in-process
	// server to connect to.
	ErrNoTarget = errors.NewSentinel("nats.no_target",
		"nats: client needs either a URL or an in-process server")

	// ErrAmbiguousTarget is returned when a client is given both a URL and an
	// in-process server. Picking one silently is the wrong kind of helpful when
	// the two lead to different brokers.
	ErrAmbiguousTarget = errors.NewSentinel("nats.ambiguous_target",
		"nats: exactly one of URL or InProcess may be set")

	// ErrNotReady is returned when an embedded server does not accept
	// connections within its ready timeout.
	ErrNotReady = errors.NewSentinel("nats.not_ready",
		"nats: server did not become ready")

	// ErrUnbounded is returned when a stream or bucket would be created with no
	// limit at all. Every JetStream limit defaults to unlimited, and unlimited
	// means "until the disk is full", which is a failure at an hour nobody
	// picked rather than a policy anybody chose.
	ErrUnbounded = errors.NewSentinel("nats.unbounded",
		"nats: at least one of max age, max messages or max bytes must be set")
)

Configuration and lifecycle errors, as sentinels so a caller can tell a misconfiguration from a broker failure.

Functions

func Reconnects

func Reconnects(n int) *int

Reconnects returns a MaxReconnects value for a settings struct.

A helper because taking the address of a literal is noise at a call site, and because the pointer exists for a reason worth naming: zero means "do not reconnect" to NATS, so it cannot also be the way a caller says "I did not set this".

Types

type ClientSettings

type ClientSettings struct {
	// URL is the server to reach. Empty is only valid when connecting to an
	// embedded server in this process through client.InProcess.
	URL string `mapstructure:"url" yaml:"url" json:"url"`

	// Name identifies this connection in the server's monitoring. Worth setting:
	// "unnamed connection" is a poor thing to find in a slow-consumer report.
	Name string `mapstructure:"name" yaml:"name" json:"name"`

	// MaxReconnects is the attempt limit.
	//
	// A pointer so that zero can mean zero. NATS reads 0 as "do not reconnect",
	// and an earlier version of this struct defaulted 0 to unlimited — so a
	// service configured to fail fast stayed alive and disconnected instead,
	// which is the opposite of what it asked for.
	//
	// Nil takes DefaultMaxReconnects, which retries forever. Negative means the
	// same explicitly.
	MaxReconnects *int `mapstructure:"max_reconnects" yaml:"max_reconnects" json:"max_reconnects"`

	// ReconnectWait is the pause between attempts. Zero takes the default.
	ReconnectWait time.Duration `mapstructure:"reconnect_wait" yaml:"reconnect_wait" json:"reconnect_wait"`

	// PendingMsgs and PendingBytes bound every subscription this client makes.
	// Zero takes the defaults above, which are far smaller than the NATS ones.
	//
	// These are the whole of this module's backpressure story: NATS sheds from
	// a full subscription queue rather than blocking the publisher, and reports
	// what it shed. A bound nobody set is a shed nobody sees.
	PendingMsgs  int `mapstructure:"pending_msgs" yaml:"pending_msgs" json:"pending_msgs"`
	PendingBytes int `mapstructure:"pending_bytes" yaml:"pending_bytes" json:"pending_bytes"`
}

ClientSettings constructs a connection.

It does not carry the credential. That arrives separately as a credentials.Source, so a configuration file can be committed and the secret resolves through the estate's ladder rather than sitting next to the host name.

func (ClientSettings) WithDefaults

func (s ClientSettings) WithDefaults() ClientSettings

WithDefaults returns a copy with the safe defaults applied.

type CredentialSource

type CredentialSource func(ctx context.Context) (string, error)

CredentialSource yields the token a connection authenticates with.

A function rather than a string so the secret is resolved at connect time rather than carried around, and so this module depends on no particular credential store. The estate's ladder — an environment variable name, a keychain reference, a literal, then a well-known variable — is composed by the calling application and handed in, which is the same shape forge.CredentialSource takes and for the same reason: precedence is the consumer's decision, not a library's.

func StaticCredential

func StaticCredential(token string) CredentialSource

StaticCredential returns a source yielding a token already in hand.

For tests and for a caller that has resolved the secret by some route this module has never heard of.

type ServerSettings

type ServerSettings struct {
	// InProcessOnly runs with no listener at all: no TCP, not even loopback.
	//
	// Clients reach it through client.InProcess, which uses an in-memory
	// net.Conn rather than dialling. This is the setting that makes a
	// single-process deployment cost nothing at the network layer while leaving
	// every publish and subscribe identical to the clustered case.
	InProcessOnly bool `mapstructure:"in_process_only" yaml:"in_process_only" json:"in_process_only"`

	// Host is the interface to bind when InProcessOnly is false. Empty binds
	// every interface, which is rarely what a service wants.
	Host string `mapstructure:"host" yaml:"host" json:"host"`

	// Port is the client listener port. Zero asks the operating system for one,
	// which is useful in tests and unhelpful in production.
	Port int `mapstructure:"port" yaml:"port" json:"port"`

	// Routes are peer servers to mesh with. Empty is a single server.
	//
	// JetStream replication is RAFT, so a cluster that intends to replicate
	// needs three servers before a quorum exists. Two is not a cluster; it is
	// two servers that cannot agree.
	Routes []string `mapstructure:"routes" yaml:"routes" json:"routes"`

	// ClusterName groups servers that route to each other. Required when Routes
	// is set, because a server with no cluster name cannot be meshed.
	ClusterName string `mapstructure:"cluster_name" yaml:"cluster_name" json:"cluster_name"`

	// ClusterPort is the port peers connect to. Distinct from Port, which is
	// for clients.
	ClusterPort int `mapstructure:"cluster_port" yaml:"cluster_port" json:"cluster_port"`

	// JetStream enables persistence, streams, key/value and object storage.
	//
	// StoreDir is required when this is set: a JetStream server with no store
	// directory writes where it pleases, and "where it pleases" survives a
	// deployment differently on every host.
	JetStream bool   `mapstructure:"jetstream" yaml:"jetstream" json:"jetstream"`
	StoreDir  string `mapstructure:"store_dir" yaml:"store_dir" json:"store_dir"`

	// MaxMemory and MaxStore bound JetStream. Zero means the server's own
	// default, which is a proportion of the host rather than a number anyone
	// chose.
	MaxMemory int64 `mapstructure:"max_memory" yaml:"max_memory" json:"max_memory"`
	MaxStore  int64 `mapstructure:"max_store" yaml:"max_store" json:"max_store"`

	// ReadyTimeout bounds how long Start waits for the server to accept
	// connections, and how long an in-process dial waits for it to finish
	// starting. Zero takes DefaultReadyTimeout.
	ReadyTimeout time.Duration `mapstructure:"ready_timeout" yaml:"ready_timeout" json:"ready_timeout"`
}

ServerSettings constructs an embedded NATS server.

Every field here has an equivalent in nats-server's own options, because an embedded server is not a lesser server — it is the same server constructed in this process. What this struct adds is a smaller surface, estate-shaped naming, and defaults chosen for a service rather than for a broker operator.

func (ServerSettings) WithDefaults

func (s ServerSettings) WithDefaults() ServerSettings

WithDefaults returns a copy with the safe defaults applied.