mq

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 13, 2026 License: Apache-2.0 Imports: 31 Imported by: 0

Documentation

Overview

Package mq is the message-queue boundary: the only package that imports NATS/JetStream (enforced by the depguard rule in .golangci.yml, Key Design Decision #20), and the only one that knows how the broker works. The rest of the process says what it wants — publish an event for a table, consume the ingest queue, park a message on the dead-letter queue, replay since a time, drop what is both written and expired — in the types below. How that maps to subjects, streams, sequences, and consumers is the implementation's (EmbeddedNATS, or ExternalNATS over an operator-owned cluster), so a broker change lands here once. The behavior below is what mqtest checks: every implementation passes its suite.

Index

Constants

View Source 
const (
	DefaultNATSSubjectPrefix = "wh"
	DefaultNATSPartitions    = 1
	DefaultNATSShards        = 32
	// MaxNATSShards bounds V: the wavehouse user's permissions name every
	// shard's durable, and each shard is a consumer (a Raft group at R3).
	MaxNATSShards             = 256
	DefaultNATSIngestConsumer = "wh-ingest"
	DefaultNATSPublishTimeout = 5 * time.Second
)

Defaults for a NATSTopology's zero fields.

View Source 
const DefaultNATSFileStore int64 = 100 << 30

DefaultNATSFileStore is the file store the generated manifests are sized for by default: the shipped Helm values' JetStream PVC.

View Source 
const EmbeddedDuplicateWindow = 2 * time.Minute

EmbeddedDuplicateWindow is how long an ingest queue remembers a WithIdempotencyKey key. It must be at least lease + ceil(lease) + 1s (2*lease + 1s for a whole-second lease), which config checks against dedupe.lease at boot. A claim left to lapse after an uncertain publish is republished once the lease ends, but the in-flight 503 tells a client to retry only after the FULL lease, so an obedient client's retry can land up to ~2*lease after the original Reserve; the +1s covers a backend (DynamoDB, for one) that rounds a claim's expiry up by as much. Only a window at least that long guarantees this queue still drops the retry's second copy.

Variables

View Source 
var EmbeddedSyncAlways = true

EmbeddedSyncAlways is NewEmbedded's SyncAlways. Only a TestMain may turn it off, before any broker starts: unit tests assert nothing across a crash, and on macOS an fsync per write is most of their run time (#617).

View Source 
var ErrConsumerMismatch = errors.New("consumer does not match its config")

ErrConsumerMismatch is CreateConsumer's answer when a consumer the broker's operator owns no longer fits the ConsumerConfig: its ack_wait is shorter, or its max_ack_pending unset.

View Source 
var ErrConsumerNotFound = errors.New("consumer not found")

ErrConsumerNotFound is returned by Purger.PurgeAcked when the named consumer does not exist (yet) on a tenant's queue.

View Source 
var ErrDeliveryEnded = errors.New("consumer delivery ended")

ErrDeliveryEnded is the error a Consumer reports on failed, wrapping the broker's reason when it gave one.

View Source 
var ErrNoDeadLetterQueue = errors.New("dead-letter queue not found")

ErrNoDeadLetterQueue is returned by DeadLetterStats.DeadLetterCounts when the tenant has no dead-letter queue of its own (nothing can have been parked for it). An implementation whose tenants share one queue returns zero counts instead. Any other failure to read it is a plain error.

View Source 
var ErrQueueFull = errors.New("ingest queue is full")

ErrQueueFull is returned by Publisher.Publish when the queue that holds the topic's tenant refuses new events because it is at a byte limit — the backpressure signal the API turns into a 503 with Retry-After. Which limits there are, and which tenants share one, is the implementation's (see Broker.SetMaxBytes). An implementation that opens a queue per tenant also returns it for a tenant whose queue it cannot open yet.

View Source 
var ErrTopology = errors.New("nats topology does not match what WaveHouse needs")

ErrTopology is what a TopologyError matches.

View Source 
var ErrUnavailable = errors.New("message queue unavailable")

ErrUnavailable is returned when the broker cannot be reached or does not answer in time — a transient failure, not a refusal, that the API turns into a 503. Retry-After is the dedupe lease, rounded up to whole seconds, when the record held a claim (so an obedient client waits out the window instead of retrying straight into it), else a flat few seconds. Only a backend whose broker is out of process returns it; the embedded one's publish failures are plain errors.

View Source 
var ErrUnitHeld = errors.New("unit held by a consumer")

ErrUnitHeld is ResetOrphaned's answer while a consumer still holds the unit, or has received from it or settled a message of it too recently to be judged gone.

View Source 
var ErrUnitsUnsupported = errors.New("consumer units not supported")

ErrUnitsUnsupported is CreateConsumer's answer to a ConsumerConfig naming units on a broker that does not implement Sharded, or naming one it does not have.

Functions

func DefaultNATSCoordBucket added in v0.2.0 

func DefaultNATSCoordBucket(prefix string) string

DefaultNATSCoordBucket is the lease bucket's name for a subject prefix, as the generated manifests name it: one per prefix, so deployments sharing a NATS account under different prefixes never contend for one lease.

func FormatStoreSize added in v0.2.0 

func FormatStoreSize(n int64) string

FormatStoreSize writes n in the largest binary unit that divides it.

func ParseStoreSize added in v0.2.0 

func ParseStoreSize(s string) (int64, error)

ParseStoreSize reads a byte count written as a Kubernetes quantity: a whole number with an optional k, M, G, T or Ki, Mi, Gi, Ti suffix (100Gi).

func WriteNATSManifests added in v0.2.0 

func WriteNATSManifests(w io.Writer, o NATSManifestOptions) error

WriteNATSManifests writes the nack CRs for the topology WaveHouse checks at boot, as one multi-document YAML stream.

func WriteNATSPermissions added in v0.2.0 

func WriteNATSPermissions(w io.Writer, t NATSTopology, jsDomain string) error

WriteNATSPermissions writes the wavehouse user's permissions under t, for a connection to JetStream domain jsDomain ("" for none, mq.nats.js_domain), as the permissions block of a NATS Helm values file's user entry.

Types

type Broker added in v0.2.0 

type Broker interface {
	Publisher
	Subscriber
	ConsumerManager
	DeadLetterer
	DeadLetterStats
	Purger
	Replayer
	// SetMaxBytes applies tenant id's byte budget (its hot-reloadable
	// mq.max_bytes_gb) to that tenant's queues — how it is split between them
	// is the implementation's — opening them if the tenant has none yet. No
	// other tenant's queues are touched. An implementation whose tenants
	// share queues may only record the budget, and say so where it does. On
	// an error the implementation restores the previous budget where it can
	// (best effort: the error says when it could not, and a canceled ctx
	// abandons the restore too), and MaxBytes keeps reporting the previous
	// budget so the next call retries.
	// MaxBytes reports the budget last applied in full for id, 0 when none
	// has been.
	SetMaxBytes(ctx context.Context, id tenant.ID, maxBytes int64) error
	MaxBytes(id tenant.ID) int64
	// Stats reports the broker counters the system gauges observe.
	Stats() (observability.MQStats, error)
}

Broker is everything the process wiring needs from the MQ: every interface above plus the lifecycle and the byte budgets. EmbeddedNATS and ExternalNATS implement it; internal/app depends on this, not on either.

type Consumer added in v0.2.0 

type Consumer interface {
	// Consume delivers each message to handler on the delivery goroutine of
	// its delivery unit — the tenant's queue, or the partition that holds
	// it: one per unit, so a tenant's messages arrive in order, one at a
	// time, while different units' arrive concurrently — handler must be
	// safe for that. A handler that blocks holds back its unit's delivery —
	// that is the backpressure the ingest worker relies on; a Sharded
	// broker's consumer keeps its hold on the unit meanwhile. About prefetch
	// messages are fetched ahead across the units together: the units when
	// delivery starts split it, and a queue joined later fetches ahead its
	// share of it at that point, at least one message each (0 = the client
	// default, per unit). The returned stop asks delivery to end and returns
	// without waiting: a handler invocation already in flight, or one for a
	// message already queued client-side, may still run after stop returns,
	// so a handler must not write to anything the caller tears down right
	// after stopping.
	//
	// Delivery can also end on its own after Consume has returned: the broker
	// or the client gives up on the consumer (it was deleted, the connection
	// closed). That is reported on failed — exactly one error, and nothing
	// once stop has been called — because no message will ever arrive to say
	// so. A caller that ignores failed waits forever on a dead consumer. A
	// queue opened later that the consumer cannot join is not that: it is
	// its tenant's failure, reported by SetMaxBytes and to the tenant's
	// publishes, and delivery from every other queue goes on.
	Consume(handler func(msg *Message), prefetch int) (stop func(), failed <-chan error, err error)
}

Consumer is a live durable consumer created by ConsumerManager.

type ConsumerConfig added in v0.2.0 

type ConsumerConfig struct {
	Durable string
	// Units limits the consumer to these units of a Sharded broker's queue,
	// by the names IngestUnits gives. Nil is every unit. A broker that does
	// not implement Sharded refuses a non-nil Units.
	Units []string
	// AckWait is the redelivery timeout: a message not acked within it is
	// delivered again.
	AckWait time.Duration
	// MaxAckPending caps unacked messages broker-side, per delivery unit (a
	// tenant's queue, or the partition that holds it): delivery from a unit
	// pauses when its unacked messages hit it (backpressure), and no other
	// unit's does.
	MaxAckPending int
	// MaxHeld, when set, caps the messages each unit of a Sharded broker's
	// queue has delivered to this process and not yet had settled: at the
	// cap the consumer fetches no more from that unit than it takes to keep
	// its hold on the unit, until one settles. It is read before every fetch, so
	// the cap may change while consuming. Nil is no cap but the broker's own.
	// A broker that does not implement Sharded ignores it.
	MaxHeld func() int
}

ConsumerConfig describes a durable, explicit-ack consumer of every event on the ingest queue, or of some of its units (see Sharded). Zero values take the broker's defaults.

type ConsumerManager added in v0.2.0 

type ConsumerManager interface {
	CreateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error)
}

ConsumerManager gives access to durable consumers on the ingest queue, held on every tenant's queue — those opened later included. Whether CreateConsumer creates the durable, or only finds one someone else made and checks it against the config, is the implementation's. A delivered Message.Ctx is the ctx given to CreateConsumer: unlike Subscriber, the consumer path does not extract the trace context carried in the message headers, because its one consumer (the ingest worker) batches across messages and never reads a per-message context.

type DeadLetterCounts added in v0.2.0 

type DeadLetterCounts struct {
	// Tables maps table name → parked messages, for the tables asked about;
	// empty, never nil, when none has any.
	// Every scope of a table counts under the table; scope is not broken out
	// yet (it is inert until #235).
	Tables map[string]uint64
	// Total is every parked message of the tenant, whatever the filter.
	Total uint64
}

DeadLetterCounts is what is parked on one tenant's dead-letter queue.

type DeadLetterStats added in v0.2.0 

type DeadLetterStats interface {
	// DeadLetterCounts counts tenant id's parked messages per table — a
	// tenant served, rejected, or removed alike, for as long as its queue is
	// kept. A non-empty table narrows Tables to that one (all of its
	// scopes). A tenant with nothing parked has zero counts, or
	// ErrNoDeadLetterQueue when it has no queue at all.
	DeadLetterCounts(ctx context.Context, id tenant.ID, table string) (DeadLetterCounts, error)
}

DeadLetterStats reports on the dead-letter queues.

type DeadLetterer added in v0.2.0 

type DeadLetterer interface {
	// DeadLetter stores msg's data on the dead-letter queue of msg's tenant,
	// under msg's topic, with the headers the options set. It does not ack
	// msg: the caller acks once the parking is confirmed, so a failure here
	// leaves the original to be redelivered.
	DeadLetter(ctx context.Context, msg *Message, opts ...PublishOpt) error
}

DeadLetterer parks messages on the dead-letter queue.

type EmbeddedNATS 

type EmbeddedNATS struct {
	// contains filtered or unexported fields
}

EmbeddedNATS runs an in-process NATS server with JetStream, and gives each tenant a queue of its own: an ingest stream and a dead-letter stream (subject.go names them), each with its own byte cap, and the durable consumers on the ingest one. Nothing outside this package sees that layout.

func NewEmbedded 

func NewEmbedded(storeDir string) (*EmbeddedNATS, error)

NewEmbedded starts an embedded NATS server with JetStream over storeDir and takes stock of the tenants' queues already there: a consumer created later is held on every one of them, those of tenants no longer served included, whose queued rows still have to reach the ingest worker. The pair of streams an earlier build kept for every tenant together is deleted, since its subjects overlap every tenant's; the events it held are not carried over. A tenant's queue is opened by SetMaxBytes, the first time its budget is applied, or by a publish or park that finds it missing, at the budget last asked for it. The server logs through slog's default logger.

func (*EmbeddedNATS) Close 

func (e *EmbeddedNATS) Close() error

func (*EmbeddedNATS) CreateConsumer added in v0.2.0 

func (e *EmbeddedNATS) CreateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error)

CreateConsumer creates or updates a durable explicit-ack pull consumer on every tenant's queue, and joins each queue opened later — one it cannot join is that tenant's failure (SetMaxBytes), not the consumer's. ctx becomes every delivered Message.Ctx (see ConsumerManager); it does not stop delivery — Consumer.Consume's stop does. It has no units: a config naming any is ErrUnitsUnsupported.

func (*EmbeddedNATS) DeadLetter added in v0.2.0 

func (e *EmbeddedNATS) DeadLetter(ctx context.Context, msg *Message, opts ...PublishOpt) error

DeadLetter stores msg's data on its topic's dead-letter subject, in its tenant's queue — the subject it arrived on with the ingest prefix swapped for the dead-letter one, nothing decoded or re-encoded. The dead-letter stream is DiscardOld, so a full one drops its oldest parked rows rather than refusing. A dead-letter stream found missing is opened again with its tenant's queue, paced as Publish's is (reopenPaced); a park refused leaves its row unacked, to be redelivered.

func (*EmbeddedNATS) DeadLetterCounts added in v0.2.0 

func (e *EmbeddedNATS) DeadLetterCounts(ctx context.Context, id tenant.ID, table string) (DeadLetterCounts, error)

DeadLetterCounts reads tenant id's dead-letter stream's per-subject counts and keys them by table (deadLetterTables). The table filter matches every scope of that table, so it is applied to the parsed topic rather than as a subject filter.

func (*EmbeddedNATS) MaxBytes added in v0.2.0 

func (e *EmbeddedNATS) MaxBytes(id tenant.ID) int64

MaxBytes reports the budget tenant id's queue was last given in full (by SetMaxBytes, or read back from disk at boot), 0 when it has none.

func (*EmbeddedNATS) Publish 

func (e *EmbeddedNATS) Publish(ctx context.Context, topic Topic, data []byte, opts ...PublishOpt) error

Publish stores data on topic's ingest subject, in its tenant's queue. A topic without a valid tenant is refused before anything is sent (see subject). A tenant with no queue has one opened at the budget last asked for it (see SetMaxBytes) — and so does one whose stream exists but whose queue the broker has not recorded open, since no consumer may hold that stream (see reopenPaced for how often a publish tries). A queue that cannot be opened — none asked for yet, JetStream refused it, or a consumer could not join it — and a queue at its byte budget (DiscardNew) are reported as ErrQueueFull: either way the tenant's queue takes nothing now, and a retry is the caller's answer.

func (*EmbeddedNATS) PurgeAcked added in v0.2.0 

func (e *EmbeddedNATS) PurgeAcked(ctx context.Context, consumer string, olderThan map[tenant.ID]time.Time) (bool, error)

PurgeAcked purges each tenant's ingest stream below MIN(consumer's ack floor + 1, first sequence stored at or after the tenant's cutoff) — see purgeAcked. A tenant olderThan does not name is purged up to its ack floor. A failure on one tenant's stream is joined into the error and the sweep goes on to the next; a done ctx ends it.

func (*EmbeddedNATS) ReplaySince added in v0.2.0 

func (e *EmbeddedNATS) ReplaySince(ctx context.Context, topic Topic, since time.Time, send func(data []byte) bool) error

ReplaySince creates an ephemeral consumer on topic's ingest subject, in its tenant's queue, starting at since (DeliverByStartTime) and drains it to send until caught up. The consumer is ack-less and expires on its own once idle. Caught up is the client's no-messages or request-timeout answer to a pull; any other pull failure (a closed connection, a deleted consumer) is returned so the caller knows the replay ended short rather than empty. A done ctx ends the drain between pulls and returns ctx's error. A topic without a valid tenant is refused like a publish (see subject).

func (*EmbeddedNATS) SetMaxBytes added in v0.2.0 

func (e *EmbeddedNATS) SetMaxBytes(ctx context.Context, id tenant.ID, maxBytes int64) error

SetMaxBytes applies tenant id's byte budget (its hot-reloadable mq.max_bytes_gb) to its queue: the ingest stream takes maxBytes and the dead-letter stream a tenth of it. A tenant with no queue yet has one opened, its dead-letter stream first, so no row is queued that could not be parked, and every registered consumer joins it. No other tenant's queue is touched.

A consumer that cannot join is this tenant's failure alone: the error says so, the queue stays closed to the tenant's publishes (ErrQueueFull), and the consumers that had not joined are tried again by the next call and by the tenant's next publish (reopenPaced). No consumer's delivery from another tenant's queue is touched.

JetStream applies a limit change to a live stream without touching its messages: growing takes effect immediately; shrinking the ingest stream below its current size makes DiscardNew refuse new publishes until the sweeper purges it back under the cap — nothing buffered is dropped. The dead-letter stream is DiscardOld, which would delete its oldest parked rows to fit a smaller cap, so it is never capped below the bytes it holds (#532): it keeps what it has, and that is logged.

The pair moves together where it can. If the dead-letter update fails after the ingest one succeeded, the ingest resize is undone so the pair stays at the previous budget, and the next call retries both. Safe in that direction — the ingest stream is DiscardNew, so shrinking it back drops nothing stored. The undo is best effort: if it fails too, the ingest stream stays at the new limit and the dead-letter one at the previous, and the error says so. On any error MaxBytes keeps reporting the previous budget, so a later call with the new budget reapplies both.

The JetStream calls are bounded by resizeTimeout, plus rollbackTimeout for the undo — or another resizeTimeout for the consumers joining a queue they do not hold yet — all rooted in ctx. That is deliberate: ctx is the process's stop context, so a reload caught mid-hook by a stop gives up — undo included — rather than holding the drain past server.shutdown_timeout. A cancellation between the two updates is therefore the one way to leave the pair split, and only for the rest of a process that is exiting: the next boot applies the adopted settings to it again.

func (*EmbeddedNATS) Stats added in v0.2.0 

func (e *EmbeddedNATS) Stats() (observability.MQStats, error)

Stats reports the embedded server's connection and inbound-message counters for observability.RegisterSystemMetrics.

func (*EmbeddedNATS) Subscribe 

func (e *EmbeddedNATS) Subscribe(ctx context.Context, consumerName string, handler func(msg *Message) error) error

Subscribe holds a durable explicit-ack consumer named consumerName on every tenant's queue, those opened later included, and delivers each message to handler with the trace context its headers carry, until ctx is done. It fetches the client's default number of messages ahead across the tenants together (see fanIn.share), so what sits client-side does not grow with the tenants. A tenant's queue that cannot be joined when it opens stays closed to the tenant's publishes until it is (SetMaxBytes), so handler misses none of its events.

type ExternalNATS added in v0.2.0 

type ExternalNATS struct {
	// contains filtered or unexported fields
}

ExternalNATS is the Broker over an operator-owned NATS cluster (see NATSTopology): every tenant shares N work-queue ingest partitions, a history stream they republish every row to, and one dead-letter stream. It never creates, changes, purges or deletes a stream or a durable. The only JetStream objects it creates are auto-expiring consumers on the history stream: one per Subscribe (the hub bridge) and one per replay.

func NewNATS added in v0.2.0 

func NewNATS(ctx context.Context, cfg NATSConfig) (*ExternalNATS, error)

NewNATS connects to the cluster, waits up to cfg.TopologyWait for the operator's topology to pass verifyNATSTopology, and returns the broker. Recommended findings are logged; a required one still missing when the wait runs out is a *TopologyError listing every finding. A cluster not reached in that time is ErrUnavailable. The topology is checked again every five minutes, reported on wavehouse_mq_topology_ok and in the log, and never repaired.

func (*ExternalNATS) CheckReplayWindows added in v0.2.0 

func (e *ExternalNATS) CheckReplayWindows(windows map[tenant.ID]time.Duration)

CheckReplayWindows warns for each tenant whose gap window is longer than the history stream keeps: SSE replay serves only the last max_age of it. It warns once per tenant and window, so calling it after every settings reload repeats nothing until a window changes. No I/O: max_age is read at boot and by the periodic poll, which runs the check again with the last windows when max_age changes.

func (*ExternalNATS) Close added in v0.2.0 

func (e *ExternalNATS) Close() error

Close stops every consumer, so none reports failed, then drains the connection so pending acks are flushed. Safe to call more than once.

func (*ExternalNATS) CreateConsumer added in v0.2.0 

func (e *ExternalNATS) CreateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error)

CreateConsumer finds the operator's shard durables — it never creates one — and checks each against cfg: its ack_wait must cover cfg.AckWait and its max_ack_pending must be set, else ErrConsumerMismatch. cfg.Units picks the units (nil: every one, the extras still draining included). A durable name that does not map to the operator's, or a configured unit whose durable or stream is gone, is ErrConsumerNotFound; an extra whose durable is gone by now is skipped.

func (*ExternalNATS) DeadLetter added in v0.2.0 

func (e *ExternalNATS) DeadLetter(ctx context.Context, msg *Message, opts ...PublishOpt) error

DeadLetter parks msg's data on the shared dead-letter stream under its topic, with a fresh Nats-Msg-Id. It does not ack msg.

func (*ExternalNATS) DeadLetterCounts added in v0.2.0 

func (e *ExternalNATS) DeadLetterCounts(ctx context.Context, id tenant.ID, table string) (DeadLetterCounts, error)

DeadLetterCounts counts tenant id's parked messages on the shared dead-letter stream, by a subject filter on its tenant: one call. Per-table counts are keyed by deadLetterTables, whose table filter matches every scope of that table. A tenant with nothing parked has zero counts; there is no queue of its own whose absence could mean anything.

func (*ExternalNATS) DuplicateWindow added in v0.2.0 

func (e *ExternalNATS) DuplicateWindow(ctx context.Context) (time.Duration, error)

DuplicateWindow is the shortest duplicate_window among the partitions: how long an idempotency key is remembered everywhere a tenant may publish.

func (*ExternalNATS) IngestUnits added in v0.2.0 

func (e *ExternalNATS) IngestUnits() (configured, extra []string)

IngestUnits implements Sharded: every partition's shard durables, then the ones outside them still draining. A unit's name is <stream>/<durable>.

func (*ExternalNATS) Leases added in v0.2.0 

func (e *ExternalNATS) Leases(ctx context.Context, bucket, holder string, opts ...LeaseOption) (coord.Coordinator, error)

Leases returns a coord.Coordinator over the operator's KV bucket on this broker's connection, holding leases as holder (the process's instance_id). A bucket that does not exist is ErrTopology: WaveHouse never creates it.

The KV revision a term was acquired at is its fencing token. Expiry is judged on the candidate's own clock, never by comparing clocks: a lease is taken over only once a candidate has seen the same revision unchanged for the lease duration. The bucket keeps no per-key TTL, since a renewal cannot extend one.

func (*ExternalNATS) MaxBytes added in v0.2.0 

func (e *ExternalNATS) MaxBytes(id tenant.ID) int64

MaxBytes reports the budget last recorded for id.

func (*ExternalNATS) Publish added in v0.2.0 

func (e *ExternalNATS) Publish(ctx context.Context, topic Topic, data []byte, opts ...PublishOpt) error

Publish stores data on topic's subject in its table's partition and shard, bounded by the topology's PublishTimeout per attempt. A publish that gets no answer is sent again up to twice with the same Nats-Msg-Id — the caller's WithIdempotencyKey when given — which the partition's duplicate window stores once. A partition at max_bytes, or a topic at its max_msgs_per_subject, is ErrQueueFull; no answer, a lost connection, or a partition stream that is gone is ErrUnavailable. It never creates anything.

func (*ExternalNATS) PurgeAcked added in v0.2.0 

func (e *ExternalNATS) PurgeAcked(_ context.Context, consumer string, _ map[tenant.ID]time.Time) (bool, error)

PurgeAcked removes nothing: the partitions delete each row once it is acknowledged, and the history keeps what its max_age allows, both the operator's. No sweeper is wired under this broker, so nothing calls it there; it is here for the Broker contract.

func (*ExternalNATS) Ready added in v0.2.0 

func (e *ExternalNATS) Ready() error

Ready reports whether a publish can get through right now: ErrUnavailable while the connection is down (the client reconnects by itself, so this is the gap between a disconnect and the reconnect, or a server that is gone for good), ErrTopology while the topology failed its last check — run every topologyRecheck, and again as soon as the client reconnects, so a cluster that came back without a stream is not ready for long. The readiness probe reads it. Under coord.backend=nats the lease bucket rides this connection and is part of the check, so it is covered too.

func (*ExternalNATS) ReplaySince added in v0.2.0 

func (e *ExternalNATS) ReplaySince(ctx context.Context, topic Topic, since time.Time, send func(data []byte) bool) error

ReplaySince reads topic's events from the history stream, stored at or after since, through an ack-less consumer of its own that expires once idle, in batches, until send returns false or the events the server counted when the replay began are sent. Anything older than the history's max_age is gone. A pull that fails before then is an error; a done ctx returns ctx's error.

func (*ExternalNATS) ResetOrphaned added in v0.2.0 

func (e *ExternalNATS) ResetOrphaned(ctx context.Context, name string) (bool, error)

ResetOrphaned implements Sharded. A pinned client holds the unit, and one that delivered or had an ack within the durable's pinned_ttl may still: its pin can be gone for a moment (a consumer leader change releases it) while it lives. Either is ErrUnitHeld. Otherwise, with rows awaiting an ack, whoever received them is gone: the reset redelivers them from the ack floor at once, where the server would wait out ack_wait. Rows already acked are not delivered again.

func (*ExternalNATS) SetMaxBytes added in v0.2.0 

func (e *ExternalNATS) SetMaxBytes(_ context.Context, id tenant.ID, maxBytes int64) error

SetMaxBytes records tenant id's budget and enforces nothing: the tenants of a partition share its max_bytes, and max_msgs_per_subject caps each topic. The first call says so in the log.

func (*ExternalNATS) Stats added in v0.2.0 

func (e *ExternalNATS) Stats() (observability.MQStats, error)

Stats reports this process's connection and its inbound message count.

func (*ExternalNATS) Subscribe added in v0.2.0 

func (e *ExternalNATS) Subscribe(ctx context.Context, consumerName string, handler func(msg *Message) error) error

Subscribe delivers every ingest event stored on the history stream from now on to handler, on one goroutine, until ctx is done or Close. It reads through an ordered ack-less consumer of its own, which skips nothing across reconnects and expires once this process is gone: every pod's hub needs every event, which one shared durable would split between them. So consumerName names nothing here, and a handler error has no redelivery to ask for: it is logged.

func (*ExternalNATS) Unowned added in v0.2.0 

func (e *ExternalNATS) Unowned(ctx context.Context) (int, error)

Unowned implements Sharded: the units whose durable has rows pending or awaiting an ack while no client holds its pin and none was active on it within its pinned_ttl. A live owner keeps its pin by pulling, even with nothing more to fetch, so such a unit has nobody writing it.

type Finding added in v0.2.0 

type Finding struct {
	Severity FindingSeverity
	// Object is what the finding is about, e.g. "stream WH_INGEST_0".
	Object string
	// Field is the setting, e.g. "retention", in the stream or consumer
	// config's own (JSON) names.
	Field string
	// Problem says what is wrong and what is needed.
	Problem string
}

Finding is one way the operator's topology differs from what WaveHouse needs.

func (Finding) String added in v0.2.0 

func (f Finding) String() string

type FindingSeverity added in v0.2.0 

type FindingSeverity int

FindingSeverity says whether a finding stops WaveHouse from serving. 

const (
	// FindingRequired refuses boot.
	FindingRequired FindingSeverity = iota
	// FindingRecommended is logged as a warning.
	FindingRecommended
)

func (FindingSeverity) String added in v0.2.0 

func (s FindingSeverity) String() string

type Halter added in v0.2.0 

type Halter interface {
	Halt()
}

Halter is implemented by a Consumer that can stop fetching ahead of its stop, so that its caller can write what it holds before the consumer lets its units go. Halt returns once no handler invocation is running and none will: everything already fetched has reached the handler. The consumer keeps its hold on its units until stop, but not indefinitely: a Sharded broker's for at least its ack wait after the halt has handed on what it fetched.

type Headers added in v0.2.0 

type Headers map[string][]string

Headers carries a message's headers. It has the same map[string][]string shape as NATS and HTTP headers, so it converts to either without a copy. Keys are exact (case-sensitive, no canonicalization), matching nats.Header.

func (Headers) Add added in v0.2.0 

func (h Headers) Add(key, value string)

Add appends value to the values associated with key.

func (Headers) Get added in v0.2.0 

func (h Headers) Get(key string) string

Get returns the first value associated with key, or "" if there is none.

func (Headers) Set added in v0.2.0 

func (h Headers) Set(key, value string)

Set replaces the values associated with key with the single value.

type LeaseOption added in v0.2.0 

type LeaseOption func(*leaseTimings)

LeaseOption adjusts a lease coordinator's timings (tests shorten them).

func WithLeaseTimings added in v0.2.0 

func WithLeaseTimings(duration, renewDeadline, renewEvery time.Duration) LeaseOption

WithLeaseTimings sets how long a candidate must see a lease unchanged before taking it over, how long a holder keeps a lease it cannot renew, and how often it renews. Each must be shorter than the one before.

type Message 

type Message struct {
	Ctx       context.Context
	Data      []byte
	Timestamp time.Time
	// contains filtered or unexported fields
}

Message represents a message received from the queue.

func NewMessage 

func NewMessage(ctx context.Context, topic Topic, data []byte, ts time.Time, doubleAck func(context.Context) error, ack func() error, nak func() error, opts ...MessageOpt) *Message

NewMessage constructs a Message with ack/nak callbacks.

func (*Message) Ack 

func (m *Message) Ack() error

Ack acknowledges the message asynchronously (fire-and-forget). Use for low-criticality consumers or high-throughput scenarios where latency is more important than a "received" confirmation from the server.

func (*Message) DoubleAck 

func (m *Message) DoubleAck(ctx context.Context) error

DoubleAck acknowledges the message synchronously, blocking until the broker confirms receipt. Use this for critical ingest paths (ClickHouse writes).

func (*Message) Nak 

func (m *Message) Nak() error

Nak negatively acknowledges the message asynchronously for redelivery. This is fire-and-forget and does not require a context.

func (*Message) NakWithDelay added in v0.2.0 

func (m *Message) NakWithDelay(delay time.Duration) error

NakWithDelay negatively acknowledges the message, asking for redelivery no sooner than delay — a retry that backs off rather than coming straight back. Fire-and-forget like Nak, which it falls back to when the message has no delayed form.

func (*Message) OnSettled added in v0.2.0 

func (m *Message) OnSettled(fn func())

OnSettled arranges for fn to run once, after the first DoubleAck, Ack, Nak or NakWithDelay, whether or not the broker confirms it: the consumer has let go of the message either way, and one it failed to settle comes back from the broker as a new delivery. A consumer handing a shard on to another process waits for it, and one that caps the messages it holds frees a slot. Call it before the message is shared with another goroutine.

func (*Message) Topic added in v0.2.0 

func (m *Message) Topic() Topic

Topic is the topic the message was published on — its tenant included, which is how the consumers learn whose event it is. It decodes the key on every call: a split and two unescapes, which the per-message paths (the hub bridge, the worker) pay once each ahead of decoding the envelope.

func (*Message) TopicKey added in v0.2.0 

func (m *Message) TopicKey() string

TopicKey is the delivered form of the topic the message was published on, undecoded: free, and opaque — what a log line names the message by.

type MessageOpt added in v0.2.0 

type MessageOpt func(*Message)

MessageOpt configures a Message beyond its required callbacks.

func WithNakDelay added in v0.2.0 

func WithNakDelay(fn func(time.Duration) error) MessageOpt

WithNakDelay gives a Message its delayed negative acknowledgement (see NakWithDelay).

type NATSConfig added in v0.2.0 

type NATSConfig struct {
	URLs []string
	// Name is the connection name the server reports; default
	// wavehouse-<hostname>.
	Name string
	// CredsFile (a user JWT and nkey seed) and NKeySeedFile are exclusive
	// with each other and with User.
	CredsFile    string
	NKeySeedFile string
	User         string
	PasswordFile string
	TLS          NATSTLS
	// JSDomain is the JetStream domain, for a leafnode or hub-and-spoke
	// deployment.
	JSDomain string
	// Topology is what the operator must have created (see NATSTopology).
	// Its PublishTimeout also bounds each publish attempt.
	Topology NATSTopology
	// ConnectTimeout bounds one dial; default 5s.
	ConnectTimeout time.Duration
	// TopologyWait is how long boot waits for the operator's topology;
	// default 60s.
	TopologyWait time.Duration
	// contains filtered or unexported fields
}

NATSConfig is how ExternalNATS reaches an operator-owned NATS cluster and what topology it expects there. Secrets are file paths only.

type NATSManifestOptions added in v0.2.0 

type NATSManifestOptions struct {
	Topology NATSTopology
	// Replicas is every stream's replica count (default 3).
	Replicas int
	// FileStore is each server's JetStream file store, max_file_store
	// (default 100 GiB, the shipped Helm values' PVC; the NATS chart sets
	// max_file_store to fileStore.maxSize, else the PVC's size). A server
	// reserves every stream replica's max_bytes it holds against it and
	// refuses a stream that does not fit (JetStream error 10047), so the
	// streams' max_bytes together must fit in it.
	FileStore int64
	// PartitionMaxBytes caps each ingest partition: the backlog of unwritten
	// rows its tenants share (default 15% of FileStore). It does not follow
	// the partition count, so lowering N (a larger partition while the
	// removed ones drain) never needs more store than before; at the defaults
	// four partitions, the history and the DLQ reserve three quarters of
	// FileStore, the quarter left being headroom for the Raft logs, the lease
	// bucket and the store's own overhead.
	PartitionMaxBytes int64
	// MaxMsgsPerSubject caps one topic's backlog in a partition (default
	// 1,000,000).
	MaxMsgsPerSubject int64
	// HistoryMaxAge is how long SSE can replay (default 15m, the seed's
	// stream.gap_window_minutes); at least the longest tenant gap window.
	HistoryMaxAge time.Duration
	// HistoryMaxBytes caps the history (default a tenth of FileStore).
	HistoryMaxBytes int64
	// DLQMaxBytes caps the dead-letter stream (default a twentieth of
	// FileStore).
	DLQMaxBytes int64
	// DLQMaxMsgsPerSubject caps one topic's parked rows (default 100,000).
	DLQMaxMsgsPerSubject int64
	// PinnedTTL is how long a shard's pin outlives its holder's last pull
	// (default 10s): at least minPinnedTTL, and under the membership lease.
	PinnedTTL time.Duration
}

NATSManifestOptions sizes the nack CRs WriteNATSManifests renders. Zero fields take the defaults below, which are starting points to tune, not recommendations for any particular load.

type NATSTLS added in v0.2.0 

type NATSTLS struct {
	CAFile         string
	CertFile       string
	KeyFile        string
	ServerName     string
	HandshakeFirst bool
}

NATSTLS is the client side of TLS to the NATS servers.

type NATSTopology added in v0.2.0 

type NATSTopology struct {
	// Prefix leads every subject: <prefix>.ingest.<p>.…, <prefix>.hist.… and
	// <prefix>.dlq.….
	Prefix string
	// Partitions is N, the number of ingest partition streams.
	Partitions int
	// Shards is V, the shards of every partition: a table's rows are all in
	// one shard, and each shard has a durable of its own.
	Shards int
	// IngestConsumer names the shard durables: shard s's on every partition
	// is <IngestConsumer>-<s> (natsShardDurable).
	IngestConsumer string
	// HistoryStream names the history stream, which holds <prefix>.hist.>.
	HistoryStream string
	// PublishTimeout bounds one publish attempt; a partition's duplicate
	// window must cover every attempt (minDuplicateWindow), so a retried
	// publish is not stored twice.
	PublishTimeout time.Duration
	// DedupeLease is how long a dedupe claim stays pending (dedupe.lease); 0
	// skips its rule. A publish whose outcome was unknown keeps its claim
	// until the lease lapses, and the client's retry is published under the
	// same idempotency key, so a partition's duplicate window must still hold
	// the first copy then (dedupeDuplicateWindow).
	DedupeLease time.Duration
	// CoordBucket is the KV bucket this process holds its leases in; empty
	// when it holds none there, and then the bucket is not checked.
	CoordBucket string
	// AckWait, MaxAckPending and Prefetch are what the ingest worker asks of
	// the durable (internal/ingest/worker.go, which imports this package).
	AckWait       time.Duration
	MaxAckPending int
	Prefetch      int
}

NATSTopology is what WaveHouse needs of an operator-owned JetStream: N ingest partition streams with work-queue retention, each split into V shards by a subject token, with one pinned-client durable per shard, and each republishing every row it stores to the history stream (limits retention, for SSE replay and the live hub) under the row's subject without the partition; one dead-letter stream; and, for coord.backend=nats, a KV bucket holding the leases (Leases). The operator creates all of it (WriteNATSManifests renders it as nack CRs); WaveHouse only checks it (verifyNATSTopology) and never repairs it.

type PublishOpt 

type PublishOpt func(Headers)

PublishOpt defines a functional option for modifying a message's headers before publishing. The map is always non-nil when an option runs.

func WithHeader 

func WithHeader(key, value string) PublishOpt

WithHeader adds a key-value pair to the message headers.

func WithIdempotencyKey added in v0.2.0 

func WithIdempotencyKey(key string) PublishOpt

WithIdempotencyKey marks a publish with key: a second publish carrying the same key within the queue's duplicate window is dropped by the broker and reported as success, so republishing an event whose first publish had an unknown outcome stores it once.

type Publisher 

type Publisher interface {
	// Publish stores data as one event on topic, in the ingest queue that
	// holds the topic's tenant. A topic without a valid tenant is refused
	// before anything is sent. ErrQueueFull when that queue refuses the event
	// at a byte limit (or, per tenant, cannot be opened yet), ErrUnavailable
	// when the broker cannot take it now.
	Publish(ctx context.Context, topic Topic, data []byte, opts ...PublishOpt) error
	Close() error
}

Publisher appends events to the ingest queue.

type Purger added in v0.2.0 

type Purger interface {
	// PurgeAcked removes, from each tenant's ingest queue, the events that
	// are BOTH acknowledged by the named durable consumer (everything before
	// its first unacked event) AND stored before that tenant's cutoff in
	// olderThan. Either bound alone keeps the event: unacked events are not
	// yet written, and recent ones are still needed for replay. A tenant
	// olderThan does not name keeps no history: everything it has
	// acknowledged goes. Reports whether anything was removed, and joins
	// each failed tenant's error — ErrConsumerNotFound for one whose queue the
	// consumer has not been created on; the other tenants' are purged all the
	// same. An implementation whose retention the broker's operator owns
	// removes nothing and reports false: either way, no unacked event is
	// removed.
	PurgeAcked(ctx context.Context, consumer string, olderThan map[tenant.ID]time.Time) (purged bool, err error)
}

Purger reclaims ingest-queue storage.

type Releaser added in v0.2.0 

type Releaser interface {
	Release(ctx context.Context) error
}

Releaser is implemented by a consumer of a Sharded broker's units. Release gives the units up so that their next owner is served at once, rather than once the broker judges this consumer gone: call it after Consume's stop, once every message delivered has been settled. A unit this consumer never held is left alone.

type Replayer added in v0.2.0 

type Replayer interface {
	// ReplaySince sends the data of every event on topic stored at or after
	// since, in order, until send returns false or the queue is caught up.
	// Running out of events is the normal end; failing to start the replay, or
	// a delivery failure before it catches up, is an error. A done ctx stops
	// the replay and returns ctx's error. A topic without a valid tenant is
	// refused as Publish refuses it.
	ReplaySince(ctx context.Context, topic Topic, since time.Time, send func(data []byte) bool) error
}

Replayer re-delivers stored events for SSE gap-fill.

type Sharded added in v0.2.0 

type Sharded interface {
	// IngestUnits names the units: the configured ones, in a fixed order the
	// same in every process, then any extra ones a lower unit count left
	// holding rows, as last read, which are drained like the others.
	IngestUnits() (configured, extra []string)
	// ResetOrphaned redelivers at once what a unit's previous owner received
	// and never settled, when no consumer holds the unit, none has received
	// from it or settled a message of it for a while, and rows are awaiting
	// a settlement; it reports whether it did, and ErrUnitHeld while a
	// consumer holds the unit or was recently active on it. A missing hold
	// alone is no proof the owner is gone. Call it only for a unit whose
	// owner the caller has other evidence is dead (the membership view), and
	// before consuming it, never while consuming it: a reset also redelivers
	// what the caller itself holds.
	ResetOrphaned(ctx context.Context, unit string) (bool, error)
	// Unowned counts the units with rows waiting that no consumer holds and
	// none has received from or settled a message of for a while: rows
	// nobody is writing.
	Unowned(ctx context.Context) (int, error)
}

Sharded is implemented by a broker whose ingest queue is split into units that one consumer at a time owns: every event of one tenant table is in one unit, and the broker lets exactly one puller receive from a unit at a time. A broker that does not implement it is one queue every consumer shares.

type Subscriber 

type Subscriber interface {
	// Subscribe registers a handler for incoming events, across every
	// tenant — those whose queues open after Subscribe included. Every event
	// published after Subscribe returns is delivered; whether earlier ones
	// are is the implementation's, and so is whether consumerName names a
	// durable consumer. The handler runs one message at a time on each
	// delivery unit — a tenant's queue, or the partition that holds it — so
	// it must be safe to call concurrently for different units. The messages
	// fetched ahead of it are a fixed number split across the units, as
	// Consumer.Consume's prefetch is, so they do not grow with the number of
	// tenants.
	//
	// CONTRACT: If the handler intends to return an error to trigger automatic
	// redelivery, it MUST NOT manually call msg.Ack() or msg.Nak() beforehand.
	//
	// CONTRACT: Symmetrically, if you explicitly call msg.Nak(), do NOT also
	// return an error — the consume loop will call Nak() again on any non-nil
	// error return.
	//
	// CONTRACT: Calling msg.DoubleAck(ctx) and then returning a non-nil error is
	// undefined behavior — the consume loop will Nak() after a successful
	// broker-confirmed Ack. Call DoubleAck, then return nil on success.
	Subscribe(ctx context.Context, consumerName string, handler func(msg *Message) error) error
	Close() error
}

Subscriber delivers every event on the ingest queue, across all tenants and topics: each tenant's in the order it was published.

type Topic added in v0.2.0 

type Topic struct {
	// Tenant is whose table it is — the settings folder the events were
	// admitted under (#583). Required: Publish refuses a topic without one,
	// so no caller falls into tenant.Default by omission.
	Tenant tenant.ID
	Table  string
	Scope  string
}

Topic addresses the events of one tenant's table, optionally narrowed to a scope within it. It is the only address the rest of the process handles; the broker's own naming (subjects, prefixes, wildcards, token encoding) is derived from it inside the implementation. Table and Scope are raw names — never pre-encoded.

type TopologyError added in v0.2.0 

type TopologyError struct {
	Findings []Finding
}

TopologyError lists every finding from the last check, required ones first.

func (*TopologyError) Error added in v0.2.0 

func (e *TopologyError) Error() string

func (*TopologyError) Unwrap added in v0.2.0 

func (e *TopologyError) Unwrap() error

Source Files

View all Source files

Directories

Path Synopsis
Package mqtest is the conformance suite for mq.Broker: the behavior the rest of the process relies on, stated once and run by every implementation from its own tests.
 Package mqtest is the conformance suite for mq.Broker: the behavior the rest of the process relies on, stated once and run by every implementation from its own tests.
Package natstest stands up NATS the way an operator deploys it for mq.backend: nats: the shipped Helm values' accounts, users and permissions (deployments/nats/values.yaml) and the shipped nack manifests (deployments/nats/jetstream.yaml).
 Package natstest stands up NATS the way an operator deploys it for mq.backend: nats: the shipped Helm values' accounts, users and permissions (deployments/nats/values.yaml) and the shipped nack manifests (deployments/nats/jetstream.yaml).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.