README
¶
messaging
A bus for the estate: services put messages on it, and the thing that carries them is somebody else's problem.
Part of the phpboyscout Go toolkit. Documentation: messaging.go.phpboyscout.uk.
A domain service knows two things: this module's interface and
go/cloudevents. It never imports a transport, so moving
the carrier from a Go channel to an in-process broker to a cluster somebody else operates changes
wiring only.
What it is, and is not
A bus is the thing services put messages on. A broker is the server that carries them, and
NATS is one. This module is the first; go/messaging-nats connects it to the second, and the
in-memory backend in this module talks to no broker at all.
The bus itself keeps nothing — no queue of its own, no replay, no state to operate. What a backend
can do beyond that it declares, in Capabilities, and is held to by the conformance suite:
at-least-once delivery, storage that survives a restart, a lease it can extend, patterns, ordering,
answering requests.
The guarantee under all of them is that every subscription is bounded and every discard is counted
and attributed, because a discarded message is gone with no signal to whoever sent it, and the
count is the only evidence it existed.
Status
Pre-1.0 and tracking its specs; versions are on the
releases page. Specified by
spec 0001,
which is IMPLEMENTED, together with the backend specs 0003 and 0005 to 0010. Request and reply is
spec 0011, and
the in-memory backend implements it.
Four provider modules implement the same conformance suite:
messaging-nats (core NATS and JetStream),
messaging-sqs (FIFO and standard),
messaging-rabbitmq (quorum queues) and
messaging-amqp (AMQP 1.0, Artemis first).
Documentation
¶
Overview ¶
Package messaging is a bus: the thing services put messages on.
A broker is the server that carries them, and NATS is one. Naming them apart is deliberate, because a system that holds both should not call them the same thing on the first day. Services publish to a bus; the bus has a backend; a backend may or may not talk to a broker, and the in-memory one does not.
What a consumer sees ¶
A domain service knows two things: this module's interface, and gitlab.com/phpboyscout/go/cloudevents. It never imports a transport, and moving the carrier from a Go channel to an in-process broker to a cluster somebody else operates changes wiring only.
The unit is a CloudEvent rather than bytes. That is what makes the swap a swap: a Go pointer handed down a channel does not survive a process boundary and a marshalled CloudEvent does, and a bus that carried bytes would push the envelope decision to every consumer, who would each answer it differently.
Sequencing ¶
Instantiation registers subjects and subscribers as intent and does not require the broker to be reachable. Start connects, creates the subscriptions, and only then signals ready. A consumer gates once:
select {
case <-bus.Ready():
// publish and consume freely from here
case <-time.After(startupBudget):
return errors.New("bus did not become ready")
}
The budget is the consumer's, because how long a service should wait for messaging before declaring itself broken is a property of that service.
What it refuses, and what it declares instead ¶
The bus itself keeps nothing: no queue of its own, no replay, no state to operate. That is a refusal rather than an omission, because a bus that gains state turns a queue somebody has to size into a component somebody has to run.
What a BACKEND can do beyond that, it declares in Capabilities and is held to by the conformance suite: at-least-once delivery, storage that survives a restart, a lease it can extend, wildcard patterns, per-subscriber ordering, replay from the oldest message it holds. The in-memory backend declares almost none of it; a broker-backed one declares what its broker earns. The guarantee under all of them is that a failure is counted.
Every subscription is bounded and every discard is counted and attributed. What happens at the bound is the backend's to offer, and a caller who asks for a policy a backend cannot honour gets a working bus plus an error naming what was dropped.
Specified by https://gitlab.com/phpboyscout/go/messaging/-/wikis/specs/0001-a-message-bus-for-the-estate
Example ¶
Example shows the whole shape: declare the subscriptions with their handlers, gate once on Ready, publish.
package main
import (
"context"
"fmt"
"time"
"gitlab.com/phpboyscout/go/cloudevents"
"gitlab.com/phpboyscout/go/messaging"
"gitlab.com/phpboyscout/go/messaging/memory"
)
func main() {
seen := make(chan string, 1)
bus, err := messaging.New(memory.New(), messaging.Settings{
Subscriptions: []messaging.SubscriptionSpec{{
Name: "greeter",
Pattern: "greeting.>",
Shape: messaging.Fanout,
Handler: func(_ context.Context, d messaging.Delivery) error {
seen <- d.Subject
return nil
},
}},
})
if err != nil {
fmt.Println("construction:", err)
return
}
ctx := context.Background()
if err := bus.Start(ctx); err != nil {
fmt.Println("start:", err)
return
}
defer func() { _ = bus.Stop(ctx) }()
// Gate once. The budget is yours, because how long a service should wait for
// messaging before declaring itself broken is a property of that service.
select {
case <-bus.Ready():
case <-time.After(5 * time.Second):
fmt.Println("the bus did not become ready")
return
}
if err := bus.Publish(ctx, "greeting.en", cloudevents.Event{
SpecVersion: "1.0",
ID: "01JQ",
Source: "urn:example:docs",
Type: "com.example.greeting.v1",
}); err != nil {
fmt.Println("publish:", err)
return
}
fmt.Println("received on", <-seen)
}
Output: received on greeting.en
Index ¶
- Constants
- Variables
- func Answer(ctx context.Context, d Delivery, reply cloudevents.Event) error
- func CopyEvent(e cloudevents.Event) cloudevents.Event
- func Match(pattern, subject string) bool
- func PatternsOverlap(a, b string) bool
- func Settle(ctx context.Context, d Delivery) error
- func SettleWith(ctx context.Context, d Delivery, s Settlement) error
- func ValidatePattern(pattern string) error
- func ValidateSubject(subject string) error
- type Backend
- type Backoff
- type Bus
- func (b *Bus) HealthCheck(name string) controls.HealthCheck
- func (b *Bus) Publish(ctx context.Context, subject string, event cloudevents.Event) error
- func (b *Bus) PublishValue(ctx context.Context, subject, eventType string, v any, opts ...PublishOption) error
- func (b *Bus) Readiness() error
- func (b *Bus) Ready() <-chan struct{}
- func (b *Bus) Request(ctx context.Context, subject string, event cloudevents.Event, ...) (Reply, error)
- func (b *Bus) RequestStats() RequestStats
- func (b *Bus) RequestStatsFor(name string) RequestStats
- func (b *Bus) RequestValue(ctx context.Context, subject, eventType string, v any, opts ...RequestOption) (Reply, error)
- func (b *Bus) Start(ctx context.Context) error
- func (b *Bus) Stats(name string) (ConsumerStats, bool)
- func (b *Bus) Status() Status
- func (b *Bus) Stop(ctx context.Context) error
- func (b *Bus) Subscribe(ctx context.Context, spec SubscriptionSpec) error
- func (b *Bus) Subscriptions() []string
- type Capabilities
- type Codec
- type ConsumerState
- type ConsumerStats
- type ConsumerStatus
- type Delivery
- type DeliveryMode
- type DeliveryOptions
- type Handler
- type JSONCodec
- type LifecycleState
- type LossReason
- type Outcome
- type Overflow
- type PublishOption
- type Reply
- type ReplyCounters
- type RequestOption
- type RequestReplier
- type RequestStats
- type Responder
- type RestartPolicy
- type Session
- type Settings
- type Settlement
- type SettlementCounters
- type Shape
- type SlowHandlerPolicy
- type Source
- type SourceFinal
- type SourceID
- type SourceSpec
- type SourceStats
- type StartPosition
- type Status
- type SubscriptionSpec
Examples ¶
Constants ¶
const DefaultMaxProcessingTime = 5 * time.Minute
DefaultMaxProcessingTime bounds ExtendLease when a subscription sets no MaxProcessingTime. Five minutes: the argument against a bound was only ever against a tight one, and every comparable implementation has one (0005 D8).
const DefaultQueueBound = 256
DefaultQueueBound is how many messages a subscription may hold when its SubscriptionSpec.Bound is zero.
That every subscription is bounded is this module's and not negotiable: an unbounded queue is not a policy; it is a decision to fail once memory is gone rather than at a number somebody chose. The number itself is the caller's (0001 D7, amended 2026-09-02). What happens when it is reached is the backend's to offer.
Variables ¶
var ErrAlreadySettled = errors.NewSentinel("messaging.already_settled",
"messaging: this delivery has already been settled")
ErrAlreadySettled is returned by a second Ack or Nack on one delivery.
A handler calling both has a bug, and the bus can see it. Spec 0005 D6 says exactly one outcome is accepted, the first one declared.
var ErrContentType = errors.NewSentinel("messaging.content_type",
"messaging: the payload is not in this codec's format")
ErrContentType is returned by Delivery.Decode when a delivery's payload is in a format the bus's codec does not produce.
It fails rather than guessing, because two services disagreeing about the wire format is exactly the case worth being loud about: a decode that half-succeeds produces a plausible value from the wrong bytes, and that is worse than a refusal.
var ErrInvalidPattern = errors.NewSentinel("messaging.invalid_pattern",
"messaging: invalid subject pattern")
ErrInvalidPattern is returned for a subject pattern this module refuses.
It is deliberately fatal at construction rather than degrading (spec 0001 D16): a subject pattern is this module's guarantee rather than a backend feature being forwarded, and the two things NATS gets wrong are both silent.
var ErrInvalidSubject = errors.NewSentinel("messaging.invalid_subject",
"messaging: invalid subject")
ErrInvalidSubject is returned for a subject that cannot be published to.
var ErrNackDelay = errors.NewSentinel("messaging.nack_delay",
"messaging: the subscription cannot delay a decline by this much")
ErrNackDelay is returned by Delivery.NackAfter for a delay the subscription cannot ask for: negative, or above its ceiling (0013 D8). The delivery is left unsettled, so the handler can still Nack.
var ErrNoDeadline = errors.NewSentinel("messaging.no_deadline",
"messaging: a request needs a deadline on its context")
ErrNoDeadline is returned by Bus.Request for a context with no deadline. A request with none, against a responder that has died, would wait for ever (0011 D4).
var ErrNoReply = errors.NewSentinel("messaging.no_reply",
"messaging: no reply before the request's context ended")
ErrNoReply is returned by Bus.Request when no reply came before the context ended. It wraps the context's error, so errors.Is distinguishes a deadline from the caller cancelling.
var ErrNoReplyAddress = errors.NewSentinel("messaging.no_reply_address",
"messaging: this delivery was published, not requested, and cannot be answered")
ErrNoReplyAddress is returned by Answer for a delivery that was published rather than requested.
var ErrNoResponders = errors.NewSentinel("messaging.no_responders",
"messaging: nothing is subscribed to answer this request")
ErrNoResponders is returned by a request when nothing is subscribed to its subject. Nothing received the request, so retrying it is safe.
var ErrNoSource = errors.NewSentinel("messaging.no_source",
"messaging: the bus was built without Settings.Source")
ErrNoSource is returned by Bus.PublishValue and Bus.RequestValue on a bus built without a Settings.Source, and refuses a Responder there, which needs one to build a failure reply.
It is refused rather than defaulted. CloudEvents makes Source half of an event's identity, so a made-up one would make two services' events collide in a way nobody would notice until they did.
var ErrNoSubscriptions = errors.NewSentinel("messaging.no_subscriptions",
"messaging: a bus needs at least one subscription")
ErrNoSubscriptions is no longer returned.
Deprecated: New accepts an empty set, because a process that only publishes or only asks still needs a bus. Kept so code that names it compiles; it will be removed before 1.0.
var ErrNotReady = errors.NewSentinel("messaging.not_ready",
"messaging: the bus has no live activation")
ErrNotReady is returned by Bus.Publish whenever no activation is live: before the first start, during a reactivation gap, and after a stop.
It is one error rather than three because the caller's response is the same, and the message is refused rather than held. "It was rejected" and "it was quietly held and delivered afterwards" look identical from the publisher's side and are opposite contracts.
It can happen after Ready() has closed ¶
Ready is a first-ready latch (0002 D8). After a health-driven restart the bus is briefly without an activation, and a publish in that window is refused. At-most-once means nobody retries on the caller's behalf; Bus.Status is how a caller tells a gap from a bus that has stopped for good.
var ErrNotSupported = errors.NewSentinel("messaging.not_supported",
"messaging: the backend does not support that")
ErrNotSupported is returned by a backend asked for a capability it does not have.
A caller treats it identically to the capability interface being absent, which is why both forms exist: a backend may decline at construction by not implementing the interface, or at call time by returning this.
var ErrQueueFull = errors.NewSentinel("messaging.queue_full",
"messaging: the subscription queue is full")
ErrQueueFull is returned to a publisher by a subscription whose overflow policy is Fail.
It is the only policy that tells the publisher anything. Under the others a discarded message is gone with no signal to whoever sent it, which is why the count is not optional.
var ErrReservedAttribute = errors.NewSentinel("messaging.reserved_attribute",
"messaging: the event carries an extension attribute this module reserves")
ErrReservedAttribute is returned for an application event carrying the extension attribute this module reserves to mark a failure reply (0011 D9).
var ErrResponderFailed = errors.NewSentinel("messaging.responder_failed",
"messaging: the responder failed to answer")
ErrResponderFailed is returned by Bus.Request when the responder returned an error or panicked. Its detail stays in the responder's process (0011 D5).
var ErrSettlementRefused = errors.NewSentinel("messaging.settlement_refused",
"messaging: the backend could not perform this settlement")
ErrSettlementRefused is returned by a backend that could not perform a settlement: the source has retired, the session has closed, or the broker could not be reached.
It is ONE sentinel because the reason does not change what the core has to do. In every case the message is unacknowledged at the backend: a persistent backend will offer it again, a non-persistent one has lost it, and the core cannot tell which without a storage capability it does not yet have. What it can do is count the refusal rather than swallow it.
var ErrSourceEnded = errors.NewSentinel("messaging.source_ended",
"messaging: the source ended itself; its link is gone")
ErrSourceEnded is returned by a subscription whose source ended itself: its transport link died rather than the core retiring it (0010 D2). The subscription fails, so the caller's RestartPolicy decides what happens next.
var ErrStopped = errors.NewSentinel("messaging.stopped",
"messaging: the bus is stopping or stopped")
ErrStopped is returned by Subscribe once shutdown has begun, or after it. Accepting a subscription that will never be supervised would send the caller away believing it has one, which is worse than an error it can see.
Bus.Request returns it for a request that Stop cancelled (0011 D3).
Functions ¶
func Answer ¶ added in v0.3.0
Answer sends reply to whoever requested d. Like Settle, the bus calls it and a handler never does; it is exported for the conformance suite, which drives a backend with no bus above it.
func CopyEvent ¶
func CopyEvent(e cloudevents.Event) cloudevents.Event
CopyEvent returns an event whose mutable parts are the caller's alone.
A CloudEvent carries a byte slice and a map, so handing the same value to every fan-out subscriber shares mutable storage between them. The natural handler shape triggers it: unmarshal into the slice, or annotate an extension, and the siblings see the edit. Between two live handlers it is also a data race, and neither is a fault a consumer could reasonably be expected to anticipate from an interface that says every subscriber receives a copy.
A backend must give each subscriber its own, and the conformance suite checks it. This is here so no backend has to work out what "its own" means.
func Match ¶
Match reports whether subject is selected by pattern.
The rules are NATS's, implemented rather than approximated. A matcher written from the documented rules agreed with a real server on all 21 rows of the spike's truth table; a prefix-and-glob approximation disagreed on 6, in both directions. See specs/reports/2026-08-29-subject-patterns.
An invalid pattern matches nothing, and so does an invalid subject. That is not defensiveness: an empty token makes a subject invalid, and the one row a from-spec matcher gets wrong until that rule is added is `a.*.c` against `a..c`, where it matches and NATS does not.
func PatternsOverlap ¶ added in v0.3.0
isPattern reports whether a valid pattern selects more than one subject. PatternsOverlap reports whether some subject is selected by both patterns. An invalid pattern overlaps nothing.
func Settle ¶
Settle performs the settlement a handler declared on d, and reports whether the backend could act on it.
This is what the bus does after a handler returns, and a handler never calls it: Ack and Nack record intent, and the bus performs the settlement so a handler is never blocked on a broker round trip mid-work (0005 D6). It is exported so that something standing in for the bus — the conformance suite, which drives a backend at the seam with no bus above it — can complete the exchange the same way the bus does. One path, not two.
It settles at most once per delivery: a second call is a no-op returning nil, because the backend has already been told.
func SettleWith ¶ added in v0.5.0
func SettleWith(ctx context.Context, d Delivery, s Settlement) error
SettleWith records s on d and performs it, as Settle performs what a handler declared. The bus never calls it: it is for driving a backend at the session with no bus above it, as the conformance suite does (0013 D4).
Recording refuses a delivery already declared with ErrAlreadySettled, and performing is a no-op if Settle has already run. A delay on an Acked settlement, or a negative one, is refused before anything is recorded.
func ValidatePattern ¶
ValidatePattern reports whether pattern may be subscribed to.
It refuses more than NATS does, on purpose. NATS accepts `a.**` and `a.>>` silently: the subscription is created, looks healthy, and can never match, because no token will ever equal `**`. And `a.>.c` is not caught client-side at all; it reaches the server, which closes the connection, so every other subscription on that connection dies with it. One malformed pattern takes down the entire bus.
func ValidateSubject ¶
ValidateSubject reports whether subject may be published to.
A subject is not a pattern: a wildcard here would publish to a name no subscriber can ever hold.
Types ¶
type Backend ¶
type Backend interface {
// Capabilities is asked before anything is built (0001 D16).
Capabilities() Capabilities
// Open connects and returns a session. It is called when the bus starts,
// not when it is built, so that a briefly unavailable broker does not stop
// an application booting.
//
// Everything it acquires must be reachable from the returned Session, or it
// leaks: that is the one obligation this contract places on an implementer,
// and it replaces the old Start/Stop pair precisely so that "what did Start
// acquire" has an answer a caller can hold.
Open(ctx context.Context) (Session, error)
}
Backend is what this module needs of a transport, and no more.
The contract carries our semantics rather than the union of every broker's. What a transport can do beyond the minimum it DECLARES, in Capabilities, and is held to: persistence across a restart, at-least-once delivery, a lease it can extend, patterns, ordering, replay. What it does not declare it is never asked for. The guarantee under all of it is that a failure is counted.
type Backoff ¶ added in v0.5.0
Backoff is a subscription's schedule for delaying declined messages (0013 D8). The delay before attempt n+1 is Initial × Multiplier^(n-1), capped at Max. The zero value is no schedule: a Nack redelivers at once.
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is the module's entry point: a service that publishes and consumes CloudEvents over a backend.
It is restartable. A stop releases everything and leaves the bus dormant rather than poisoned, so a Controller's health policy can bounce it.
func New ¶
New validates the settings and returns a bus that has connected to nothing.
Construction is offline on purpose: a briefly unavailable broker must not stop an application booting.
It can return a usable bus AND an error ¶
Something wrong with the caller's own declaration — no handler, no name, a bad pattern, a duplicate — is fatal, and the bus is nil. A setting this backend cannot honour degrades instead: the bus is usable, the setting is not applied rather than approximated, and the error names every dropped setting rather than the first (0001 D16).
So the test is bus == nil, not err != nil.
Example (Degraded) ¶
ExampleNew_degraded shows a construction that returns a usable bus AND an error. The test is bus == nil, not err != nil.
package main
import (
"context"
"fmt"
"gitlab.com/phpboyscout/go/messaging"
"gitlab.com/phpboyscout/go/messaging/memory"
)
func main() {
// A backend offering only its own native behaviour. Core NATS is like this:
// it can shed the newest and nothing else, so a caller moving from the
// in-memory backend to it must be told what it lost rather than quietly
// getting something else.
bus, err := messaging.New(nativeOnlyBackend{}, messaging.Settings{
Subscriptions: []messaging.SubscriptionSpec{{
Name: "worker",
Pattern: "work.>",
Shape: messaging.Fanout,
Overflow: messaging.ShedOldest,
Handler: func(context.Context, messaging.Delivery) error { return nil },
}},
})
if bus == nil {
fmt.Println("fatal:", err)
return
}
fmt.Println("usable:", bus != nil)
fmt.Println("degraded:", err != nil)
}
// nativeOnlyBackend declares no overflow policy of its own, so any request for
// one degrades.
type nativeOnlyBackend struct{}
func (nativeOnlyBackend) Capabilities() messaging.Capabilities {
return messaging.Capabilities{Patterns: true, DeliveryModes: []messaging.DeliveryMode{messaging.AtMostOnce}}
}
func (nativeOnlyBackend) Open(ctx context.Context) (messaging.Session, error) {
return memory.New().Open(ctx)
}
Output: usable: true degraded: true
func (*Bus) HealthCheck ¶
func (b *Bus) HealthCheck(name string) controls.HealthCheck
HealthCheck reports degraded while any subscription has terminally failed, naming them.
It reads the board rather than a live supervisor, because the supervisor that knew a terminal failure is discarded with its activation, and a fact that dies with its witness is not a record.
func (*Bus) Publish ¶
Publish sends one event to one subject.
It borrows the live activation for the duration of the call, so the session underneath it cannot be closed while the publish is in flight, and a publish that arrives with no activation is refused with ErrNotReady rather than landing on a half-torn-down one.
func (*Bus) PublishValue ¶
func (b *Bus) PublishValue( ctx context.Context, subject, eventType string, v any, opts ...PublishOption, ) error
PublishValue encodes v with the bus's codec and publishes it.
eventType is explicit because it is the discriminator that lets one bus carry many kinds of message without a union type, and inferring it from a Go type would tie the wire format to a package path.
Bus.Publish is unchanged and remains the way to publish an event a caller has built themselves.
func (*Bus) Readiness ¶
Readiness reports nil exactly while an activation is installed. It is the probe a Controller registers.
A terminally failed subscription does NOT make the bus unready: one consumer's defect must not take its siblings out of rotation (0001 D19). It shows as degraded through Bus.HealthCheck instead.
func (*Bus) Ready ¶
func (b *Bus) Ready() <-chan struct{}
Ready closes once the bus has been ready, and never reopens.
It answers "was this bus ever ready", which is the boot-ordering question 0001 D2 asked it and the only one a latch can answer. For current truth, including during a reactivation gap, use Bus.Status.
func (*Bus) Request ¶ added in v0.3.0
func (b *Bus) Request( ctx context.Context, subject string, event cloudevents.Event, opts ...RequestOption, ) (Reply, error)
Request sends event to subject and returns one reply, within the deadline ctx carries (0011).
An answer the caller did not want is still a reply: a refusal belongs in the reply's payload. The errors are for the exchange failing: ErrNoResponders, ErrNoReply, ErrResponderFailed, ErrStopped when Stop cancels it, or the transport's own. Refused before sending, and not counted: ErrNoDeadline, ErrNotReady, ErrNotSupported, ErrReservedAttribute.
func (*Bus) RequestStats ¶ added in v0.3.0
func (b *Bus) RequestStats() RequestStats
RequestStats is the total across every request name, with Unclaimed.
func (*Bus) RequestStatsFor ¶ added in v0.3.0
func (b *Bus) RequestStatsFor(name string) RequestStats
RequestStatsFor is one name's row; "" is requests sent without a name, and a name never declared reads as zero.
func (*Bus) RequestValue ¶ added in v0.3.0
func (b *Bus) RequestValue( ctx context.Context, subject, eventType string, v any, opts ...RequestOption, ) (Reply, error)
RequestValue encodes v with the bus's codec and requests it, as Bus.PublishValue publishes.
func (*Bus) Start ¶
Start connects the backend, establishes every subscription, and only then signals ready.
func (*Bus) Stats ¶
func (b *Bus) Stats(name string) (ConsumerStats, bool)
Stats is one consumer's accounting, cumulative for the life of the bus.
The live source's own reading is folded in, because a running source holds numbers that exist nowhere else until it retires.
func (*Bus) Status ¶
Status is a coherent reading of the bus, with a channel that closes on the next change.
State and Changed are captured together, which is what lets a caller wait for a transition without missing one:
for {
st := bus.Status()
if wanted(st) {
break
}
select {
case <-st.Changed:
case <-ctx.Done():
return ctx.Err()
}
}
func (*Bus) Stop ¶
Stop releases the activation and leaves the bus dormant.
It is bounded: it returns within the caller's budget whatever a handler does. What it does not promise is that every handler has returned — a handler that ignores cancellation is disowned rather than waited for, because Go cannot end a goroutine. Its delivery stays counted as in-hand, which is truthful, and its consumer alone is quarantined when the bus next starts.
func (*Bus) Subscribe ¶
func (b *Bus) Subscribe(ctx context.Context, spec SubscriptionSpec) error
Subscribe attaches a subscription after construction (0001 D5, OQ6): for a set not knowable at wiring, or to take a bigger share of a burst, since a group distributes per subscription rather than per worker.
Validated exactly as at construction, and refused for the same reasons, with one difference: a setting this backend does not offer is refused here rather than degraded, because a caller adding one subscription can act on an error and a bus under construction could not. Before Start the subscription is recorded and built with the rest; while running it is attached to the live generation and supervised identically, and this returns once its first subscribe has succeeded or failed; once shutdown has begun it is refused with ErrStopped. Ready still means the wiring-time set (D5).
func (*Bus) Subscriptions ¶
Subscriptions is every subscription name this bus was built with.
type Capabilities ¶
type Capabilities struct {
// Overflow lists the policies this backend can honour. A backend that
// offers none is saying every subscription gets its native behaviour.
Overflow []Overflow
// Patterns reports whether the backend can subscribe to a wildcard pattern.
// A backend with a flat namespace, Kafka topics or SQS, cannot.
Patterns bool
// DeliveryModes lists every mode this backend offers, INCLUDING AtMostOnce.
// An empty list offers nothing.
//
// AtMostOnce is not implied. SQS and Pub/Sub cannot promise never-
// duplicated delivery, so "every backend can do at-most-once" is false and
// an empty set cannot be read as meaning it. A backend declares every mode
// it offers, including the zero-value one (0005 D11).
DeliveryModes []DeliveryMode
// SurvivesRestart reports whether an unacknowledged message outlives the
// process. A STORAGE property, on its own axis, separate from the modes:
// a backend can retry until acknowledged and still lose its backlog on a
// bounce, and a subscription needs to be able to ask which (0005 D11, D13).
SurvivesRestart bool
// ExtendsLease reports whether the backend can keep a message assigned to
// a handler that is still working (0005 D7, D11). It matters only where a
// lease exists, which is only where a message can outlive the process
// holding it: a backend that survives a restart and cannot extend refuses
// the default SlowHandler policy and requires Redeliver. A backend with no
// lease — one that never hands a message on while a handler runs — has
// nothing to refuse and declares false honestly.
ExtendsLease bool
// Ordered reports whether one subscriber receives one subject's messages in
// publication order. Memory and a NATS subscription do; a standard SQS
// queue does not; a FIFO one does, at the price of one message in flight
// per subject, which is why the two are one declaration: an ordered
// backend is held to the ordering row, and the per-message-acknowledgement
// row uses two subjects on it rather than assuming two in flight on one.
Ordered bool
// Replays reports whether a NEW durable identity can begin at the oldest
// message the backend still retains: a log with a cursor, which JetStream
// is, rather than a queue, which SQS, RabbitMQ and AMQP are. StartAt is
// meaningful only here; AllAvailable on a backend without it is refused at
// construction, because a queue delivers what it holds and "from the
// start" would be a promise nobody keeps.
Replays bool
// RestartLosesLastFailure reports whether a restart can bring an
// unacknowledged message back numbered one lower than the attempts it has
// actually had.
//
// A backend that survives a restart keeps the attempt count with the
// message, so MaxDeliver bounds a MESSAGE rather than one process's memory
// of it. A transport that writes the count as it delivers, rather than as
// each attempt fails, holds the last failure in memory only: Artemis does,
// and an orderly restart there returns a message numbered for the
// deliveries it made, not the failures it saw. The budget is still bounded
// and the count still does not reset, so this is declared and held to
// rather than made a lie of SurvivesRestart (0005 D12 row 2, 0008 finding
// 16).
//
// Meaningless without SurvivesRestart, and false on every backend that
// persists a failure when it happens.
RestartLosesLastFailure bool
// RestartCanSpendUnseenAttempt reports whether a restart can spend an
// attempt that no process saw. When that attempt was the last, the message
// is exhausted and no source counts it Exhausted.
//
// JetStream does: a redelivery that races a consumer's unsubscribe can be
// counted as sent and reach nobody, and a server restarted before the next
// pull never offers that attempt again (0012). Meaningless without
// SurvivesRestart.
RestartCanSpendUnseenAttempt bool
// DelaysRedelivery reports whether a declined message can be held back
// before its next attempt, never offered again before its delay has passed
// (0013 D1). A backend that declares it must set
// DeliveryOptions.OnSettleWith.
DelaysRedelivery bool
// MaxRedeliveryDelay is the longest delay the backend can hold a message
// for. Zero, with DelaysRedelivery, means no limit.
MaxRedeliveryDelay time.Duration
// RoutesAroundDegraded reports whether a competing group avoids a member
// whose queue is full.
//
// Almost always false, and it is here because it is the difference a
// consumer is most likely to assume the other way: "queue group" suggests
// load balancing and NATS does not do it.
RoutesAroundDegraded bool
// RequestReply reports whether the backend answers requests: its sessions
// implement [RequestReplier], and its deliveries carry a reply address when
// one was sent (0011 D3).
RequestReply bool
}
Capabilities is what a backend says about itself before it is asked to do anything.
Declared rather than discovered by trying, because construction has to degrade honestly (spec 0001 D16) and cannot do that by attempting an operation and seeing what happens.
func (Capabilities) Supports ¶
func (c Capabilities) Supports(o Overflow) bool
Supports reports whether the backend declared this overflow policy. OverflowNative is always supported: it is the absence of a request.
func (Capabilities) SupportsMode ¶
func (c Capabilities) SupportsMode(m DeliveryMode) bool
SupportsMode reports whether the backend declared this delivery mode. Nothing is implied by an empty list, not even AtMostOnce.
type Codec ¶
type Codec interface {
// ContentType is the media type this codec produces, and is what lands in
// an event's DataContentType.
//
// It belongs on the codec rather than beside it: a codec that does not know
// what it produces cannot label its own output, and an unlabelled payload
// is one a second service cannot read.
ContentType() string
// Encode turns a value into a payload.
Encode(v any) ([]byte, error)
// Decode fills into from a payload. into is a pointer.
Decode(data []byte, into any) error
}
Codec turns a value into a CloudEvent's payload and back.
Why this exists ¶
A CloudEvent already carries DataContentType and DataSchema, so Protobuf, Avro, MessagePack, CBOR and raw JSON all fit inside the envelope as it stands. What was missing was not permission but ownership: the struct-to-bytes step had none, so every consumer marshalled at its own call sites and chose its own schema. Two services then answer that differently while both believing they use the same bus — which is the argument for carrying a CloudEvent at all, reproduced one level down inside the payload where that decision does not reach.
See spec 0004.
type ConsumerState ¶
type ConsumerState uint8
ConsumerState is one subscription's lifecycle fact.
const ( // ConsumerUnknown is a consumer the bus has never run. ConsumerUnknown ConsumerState = iota // ConsumerRunning has a live source. ConsumerRunning // ConsumerQuarantined is waiting for a previous generation's handlers to // return, and deliberately has no source while it waits (0002 D7). ConsumerQuarantined // ConsumerStopped ended cleanly. ConsumerStopped // ConsumerFailed exhausted its restart policy. ConsumerFailed )
func (ConsumerState) String ¶
func (s ConsumerState) String() string
String makes the state loggable.
type ConsumerStats ¶
type ConsumerStats struct {
// Name is the subscription this describes.
Name string
// Generation is the bus activation that owns the current source, or zero
// when the consumer has none.
Generation uint64
// Offered is everything the bus was asked to give this consumer,
// cumulative across every source generation: what entered its queues plus
// what was refused at the bound.
Offered uint64
// Delivered is everything a handler returned from, successfully or with an
// error. An error is an outcome, not a loss.
Delivered uint64
// Pending is what is waiting in the current source's queue. It is a gauge
// of now, not a total, and adding it up over time means nothing.
Pending uint64
// InHand is deliveries whose handler has not returned. It is truthful
// rather than convenient: a delivery held by a handler that ignores
// cancellation stays here for ever, which is the honest answer.
InHand uint64
// Lost is why messages went missing, keyed by reason. It is a copy: the
// live map is never handed out, because sharing its backing store with a
// concurrent writer would be a data race on the very type that exists to
// prove nothing was lost quietly.
Lost map[LossReason]uint64
// Released is attempts this consumer gave up before any handler returned,
// and which the backend may hand out again. Always zero unless the
// subscription is AtLeastOnce (0006 D3). Attempt-grain: one message can be
// released more than once, so this is not a message count.
Released uint64
// Exhausted is MESSAGES that ran out of delivery attempts. Message grain,
// deliberately NOT a term in the attempt identity (0006 D2, D8).
Exhausted uint64
// LeaseExpired is ATTEMPTS whose lease reached MaxProcessingTime and were
// handed on while the handler may still have been running (0005 D8). Not a
// term in the identity: the attempt is Delivered when its handler returns.
// It says a duplicate is now possible, which is what an operator needs to
// see and cannot otherwise. Zero on a backend with no lease.
LeaseExpired uint64
// DelayShortened is declines the backend redelivered sooner than their
// delay asked, because it could not hold them that long (0013 D9). Not a
// term in the identity, and not a loss: the attempt was made, early.
DelayShortened uint64
// Stranded is MESSAGES whose last attempt declined them and whose backend
// could not be asked to try again (0006 D8). Message grain, like
// Exhausted: the attempt WAS delivered, so counting the message as lost at
// attempt grain would give Offered 1 against Delivered 1 + Lost 1. Counted
// only on a backend that does not survive a restart: on one that does, the
// lease expires and the broker offers the message again, so nothing is
// stranded and the refusal is recorded in Settlement.Refused alone.
Stranded uint64
// Redelivered is attempts that arrived with a backend attempt number above
// one. A repeat is what the backend says it is: a first delivery to this
// process carrying attempt four IS a repeat, and counting it as new
// because this process had not seen it would make the number mean
// "repeats since the last deploy". Offered - Redelivered is "backend-first
// attempts observed by this process", and nothing more.
Redelivered uint64
// Settlement is what handlers SAID about their attempts, which is a
// different grain from where the attempts went.
Settlement SettlementCounters
// Replies is what became of a responder's replies. A responder's
// deliveries are never settled, so Settlement stays zero for it.
Replies ReplyCounters
// Errors is handlers that returned an error. Counted separately from Lost
// because the message was delivered.
Errors uint64
// Panics is handlers that panicked.
Panics uint64
// Incidents is terminal consumer failures. Not a delivery count.
Incidents uint64
// UnsubscribedFor is how long this consumer has had no source at all,
// summed across restarts and quarantines. Messages published into that
// window reach nobody and cannot be counted by anything, so the honest
// report is the size of the hole rather than a fabricated number of
// messages (0002 D2).
UnsubscribedFor time.Duration
}
ConsumerStats is one consumer's accounting, for the life of the bus rather than the life of a source.
The numbers are related by an identity the suite asserts on every exit path:
Offered == Delivered + Lost(total) + InHand
Offered, not Accepted ¶
The seam's SourceStats.Accepted means "entered the queue", and a message shed at the bound never did. Reusing that word here for "everything the bus was asked to give this consumer" would be one term meaning two things at two layers, which is the disease this spec exists to remove. So the bus counts what it was Offered, and a shed message is offered-and-lost rather than never-mentioned.
Incidents and UnsubscribedFor are deliberately NOT delivery counts. One terminal failure is not one lost message, and a window with no subscription is a duration nobody can convert into a number of messages honestly.
func (ConsumerStats) TotalLost ¶
func (s ConsumerStats) TotalLost() uint64
TotalLost is every loss, whatever the reason.
type ConsumerStatus ¶
type ConsumerStatus struct {
Name string
// State is the last fact recorded for this consumer.
State ConsumerState
// Generation is the activation that recorded it.
Generation uint64
// Cause is why it failed, when it did.
Cause error
}
ConsumerStatus is what the bus knows about one subscription, and it outlives the activation that produced it.
type Delivery ¶
type Delivery struct {
// Subject is the concrete subject published to, not the pattern that
// selected it. A subscriber matching `a.>` needs to know which `a.b` it got.
Subject string
// Event is the message.
Event cloudevents.Event
// Attempt is how many times THE BACKEND has delivered this message,
// starting at one. It is the backend's durable number, not this process's:
// after a restart a message on its fourth attempt arrives with Attempt 4
// and is the first thing this process has been offered for it. A backend
// that does not count attempts reports 1; see DeliveryOptions.
Attempt int
// contains filtered or unexported fields
}
Delivery is one message crossing the boundary from a backend to this module.
func NewDelivery ¶
func NewDelivery(o DeliveryOptions) Delivery
NewDelivery builds a delivery for a backend to hand to the core.
It exists because settlement state is shared across copies of a Delivery and therefore cannot be a public field: a backend must not be able to build a Delivery whose copies disagree.
func (Delivery) Ack ¶
Ack accepts this delivery.
Under AtLeastOnce the backend acts on it and the message is gone. Under AtMostOnce there is nothing to act on, because the backend settled the message before the handler saw it. Either way the outcome is RECORDED in the ledger's Settlement counters, and a second call is refused: a handler calling Ack twice has a bug in every mode, and a rule that only bites in one is one people discover in production.
func (Delivery) Decode ¶
Decode fills into from this delivery's payload, using the bus's codec.
into is a pointer. The handler names its target type because only the handler knows it; the bus supplies the codec, so the two halves cannot drift apart.
A payload whose content type is not the codec's returns ErrContentType rather than being decoded anyway. Two services disagreeing about the format is the case worth failing loudly on: a decode that half-succeeds produces a plausible value from the wrong bytes.
func (Delivery) Nack ¶
Nack declines this delivery, which under AtLeastOnce returns it to the backend as a spent attempt: it is redelivered until MaxDeliver, after which it is exhausted. A subscription with a Backoff waits the schedule's delay before the next attempt (0013 D8). Under AtMostOnce it is recorded and acts on nothing.
func (Delivery) NackAfter ¶ added in v0.5.0
NackAfter declines this delivery and asks for its next attempt to wait at least delay, overriding the subscription's Backoff for this decline (0013 D8). On a subscription that sets neither Backoff nor MaxNackDelay it returns ErrNotSupported; for a negative delay, or one above the ceiling, it returns ErrNackDelay. Either way nothing is recorded and the handler can still Nack.
type DeliveryMode ¶
type DeliveryMode uint8
DeliveryMode is how many times a subscription's messages may be attempted, and whether the bus waits to learn the outcome.
It is DeliveryMode rather than Delivery because Delivery is already the message envelope. Spec 0005 D11 proposed `DeliveryModes []Delivery`, which does not compile.
const ( // AtMostOnce is one attempt, no acknowledgement wait, no retry. Today's // behaviour and the zero value. AtMostOnce DeliveryMode = iota // AtMostOnceConfirmed is reserved and inert: construction refuses it. // // Deferred by 0005 D2. Its promise — the caller learns whether one attempt // landed — has no observer on this side of the seam, because the process // that wants to know is the process that ran the handler. The name is // held so that a consumer cannot select it and later find its meaning // changed. When a settlement-result surface exists it is AtLeastOnce with // MaxDeliver 1, and this value may be given meaning then. AtMostOnceConfirmed // AtLeastOnce retries until the handler acknowledges, bounded by // MaxDeliver, under a DurableName. // // It does NOT imply persistence. Retrying until acknowledged is a property // of this subscription's relationship with its handler; surviving a restart // is a property of the backend's storage, which Capabilities.SurvivesRestart // reports separately. A subscription that asks for this mode on a backend // that does not survive a restart is refused unless it says AllowVolatile. // // The in-memory backend declares it, volatile; the provider modules // declare it over storage that survives — JetStream, SQS, RabbitMQ quorum // queues, and AMQP 1.0 where the broker's configuration earns it. AtLeastOnce )
func (DeliveryMode) String ¶
func (m DeliveryMode) String() string
type DeliveryOptions ¶
type DeliveryOptions struct {
Subject string
Event cloudevents.Event
// Attempt is the backend's delivery count for this message, starting at
// one. Zero is read as one, so a backend that does not count attempts
// need not say so.
Attempt int
// OnSettle is called exactly once, after the handler returns, with the
// outcome the handler declared — Silent if it said nothing — and reports
// whether the backend could act on it. Nil means this backend offers no
// settlement, which is what AtMostOnce needs.
OnSettle func(context.Context, Outcome) error
// OnSettleWith is OnSettle with room to grow, and carries a decline's
// delay. When it is set the bus calls it instead of OnSettle. A backend
// that declares Capabilities.DelaysRedelivery must set it (0013 D4).
OnSettleWith func(context.Context, Settlement) error
// OnReply sends a reply to whoever requested this delivery. Nil means it
// was published, not requested, and has nowhere to answer (0011 D3).
OnReply func(context.Context, cloudevents.Event) error
}
DeliveryOptions is what a backend supplies to build one delivery.
A struct rather than positional parameters, because this constructor is public API at a seam that has already grown twice — settlement, then the attempt number — and an adapter in another module compiled against a positional signature breaks on every addition. A review predicted exactly this and it happened before the slice was a day old.
type Handler ¶
Handler processes one delivery.
Under AtLeastOnce a handler MUST settle the delivery, with Delivery.Ack or Delivery.Nack. Returning without settling is Silent, and the safe reading of "said nothing" is "did not accept" (0005 D5), so the message is redelivered, after the subscription's Backoff as a Nack is, until MaxDeliver and then exhausted. The returned error is not a settlement: a handler that returns nil without acking has declined the message, whatever it meant.
It returns an error, and errors are counted and logged. An error does not stop the consumer: it is a known outcome the handler chose to report, where a panic is an unknown state (0002 OQ2a). There is no path in this module where a panic is an expected outcome or a signalling mechanism.
Example ¶
ExampleHandler shows the one rule an at-least-once handler must follow: settle the delivery. A handler that returns without acking has declined the message, whatever it meant, and the backend redelivers it until MaxDeliver.
The in-memory backend keeps nothing across a restart, so a durable subscription on it says AllowVolatile; a broker-backed one does not need to.
package main
import (
"context"
"fmt"
"time"
"gitlab.com/phpboyscout/go/cloudevents"
"gitlab.com/phpboyscout/go/messaging"
"gitlab.com/phpboyscout/go/messaging/memory"
)
func main() {
done := make(chan struct{})
bus, err := messaging.New(memory.New(), messaging.Settings{
Subscriptions: []messaging.SubscriptionSpec{{
Name: "invoices",
Pattern: "invoice.raised",
Shape: messaging.Fanout,
Mode: messaging.AtLeastOnce,
DurableName: "invoices",
MaxDeliver: 5,
AllowVolatile: true,
Handler: func(_ context.Context, d messaging.Delivery) error {
if err := process(d); err != nil {
// Declined: the backend redelivers it as a spent attempt.
_ = d.Nack()
return err
}
close(done)
// Accepted: the backend is told, and the message is gone.
return d.Ack()
},
}},
})
if err != nil {
fmt.Println("construction:", err)
return
}
ctx := context.Background()
if err := bus.Start(ctx); err != nil {
fmt.Println("start:", err)
return
}
defer func() { _ = bus.Stop(ctx) }()
select {
case <-bus.Ready():
case <-time.After(5 * time.Second):
fmt.Println("the bus did not become ready")
return
}
if err := bus.Publish(ctx, "invoice.raised", cloudevents.Event{
SpecVersion: "1.0",
ID: "01JR",
Source: "urn:example:docs",
Type: "com.example.invoice.raised.v1",
}); err != nil {
fmt.Println("publish:", err)
return
}
<-done
fmt.Println("settled")
}
func process(messaging.Delivery) error { return nil }
Output: settled
type JSONCodec ¶
type JSONCodec struct{}
JSONCodec is the default: encoding/json, and the media type to say so.
It is the default because it is the safe choice for most callers, because it is what this module already assumed in practice, and because a consumer who never thinks about encoding still gets a labelled, readable payload. Choosing anything else is then an explicit act by the consumer that wants it.
func (JSONCodec) ContentType ¶
ContentType is application/json.
type LifecycleState ¶
type LifecycleState uint8
LifecycleState is where the bus is. It is observable because a caller gating on readiness needs to tell a reactivation gap from a bus that never started, and those look identical from a publish error alone.
const ( // StateDormant means no activation owns resources. Start is allowed. StateDormant LifecycleState = iota // StateStarting means an activation is opening its session and // establishing its sources. Publishing is refused. StateStarting // StateRunning means every required first source exists and publish // admission is open. StateRunning // StateStopping means admission is closed and cleanup is incomplete. StateStopping )
func (LifecycleState) String ¶
func (s LifecycleState) String() string
String makes the state loggable without a lookup table at every call site.
type LossReason ¶
type LossReason string
LossReason says why a message was lost. It is an open vocabulary rather than a closed one, because a new policy or a new failure path should add a reason instead of hiding inside an existing one.
const ( // LossOverflow is a message refused at a queue's bound. LossOverflow LossReason = "overflow" // LossRetiredQueued is a backlog abandoned when a source retired. It names // what actually happened: the messages were accepted, then the generation // holding them ended. LossRetiredQueued LossReason = "retired_queued" // LossShutdown is a delivery in a handler's hand when the shutdown budget // expired. LossShutdown LossReason = "shutdown" // LossHandlerPanic is a delivery whose handler panicked. LossHandlerPanic LossReason = "handler_panic" )
type Outcome ¶
type Outcome uint32
Outcome is what a handler said about one delivery.
An adapter must treat an unrecognised Outcome as Nacked. Spec 0005 leaves room for a terminal refusal, the way JetStream's Term and Rabbit's Reject(false) let a handler say "poison, never retry", and this rule is what keeps adding one additive: an adapter compiled against today's set meets tomorrow's value by declining the message rather than accepting it.
It is uint32 rather than uint8 because the settlement state it is read back from is an atomic.Uint32, and a narrowing conversion on that read is one the linter rightly refuses. The values only ever come from this package, so the width is a statement about the storage, not the range.
type Overflow ¶
type Overflow string
Overflow is what a subscription does when its queue is full.
The bound is this module's and is not configurable; the policy at the bound is the backend's to offer. A backend declares which of these it supports and a caller asking for another gets the backend's native behaviour plus a construction error naming what was dropped.
const ( // OverflowNative leaves the policy to the backend. Asking for nothing is // never an error. OverflowNative Overflow = "" // ShedNewest refuses the arriving message and leaves the backlog intact. // The only policy core NATS can offer. ShedNewest Overflow = "shed-newest" // ShedOldest evicts the head and enqueues the new message. ShedOldest Overflow = "shed-oldest" // Fail returns an error to the publisher. Fail Overflow = "fail" )
type PublishOption ¶
type PublishOption func(*publishOpts)
PublishOption adjusts one publish. Each exists because a caller needed it rather than for symmetry: an option set filled out for completeness is surface nobody asked for.
func WithDataSchema ¶
func WithDataSchema(urn string) PublishOption
WithDataSchema sets the event's DataSchema.
The codec knows the format; only the caller knows which schema their payload conforms to, so the seam does not invent one. The estate's convention is a URN — see go/cloudevents/urn.
func WithEventID ¶
func WithEventID(id string) PublishOption
WithEventID sets the event's ID instead of letting the bus generate one.
The bus generates a UUIDv7 when nobody says otherwise, which costs an inexperienced caller nothing. This exists so that a publisher who needs IDs to MEAN something — deduplication, correlation, an idempotency key they own — has a supported path from the first release, rather than the bus owning identity and that being unwindable only by a breaking change.
type Reply ¶ added in v0.3.0
type Reply struct {
Subject string
Event cloudevents.Event
// contains filtered or unexported fields
}
Reply is the answer to one request.
type ReplyCounters ¶ added in v0.3.0
type ReplyCounters struct {
// Replied is replies the backend accepted.
Replied uint64
// Refused is replies the backend refused: the session had closed, or the
// publish failed.
Refused uint64
// Unaddressed is replies discarded because the delivery was published, not
// requested, and had nowhere to answer.
Unaddressed uint64
}
ReplyCounters is what became of a responder's replies (0011 D7). Their sum is every responder invocation that completed or panicked, because each produces exactly one reply outcome.
type RequestOption ¶ added in v0.3.0
type RequestOption func(*requestOpts)
RequestOption adjusts one request.
func WithRequestName ¶ added in v0.3.0
func WithRequestName(name string) RequestOption
WithRequestName files the request's counters under name, which must be one of Settings.RequestNames. An undeclared name is filed with the unnamed requests and logged once (0011 D8).
type RequestReplier ¶
type RequestReplier interface {
// Request sends event and returns the first reply, [ErrNoResponders] when
// nothing is subscribed to subject, or ctx.Err(). Any other error is a
// transport failure, including the one a closing session ends a waiting
// request with.
Request(ctx context.Context, subject string, event cloudevents.Event) (cloudevents.Event, error)
// Unclaimed is how many replies arrived for a request that had already
// finished. It must not block, and stays readable after Close, because the
// bus reads it under its ledger's lock (0011 D7).
Unclaimed() uint64
}
RequestReplier is implemented by a Session whose backend declares Capabilities.RequestReply (0011 D3).
type RequestStats ¶ added in v0.3.0
type RequestStats struct {
// Sent is every request handed to the backend.
Sent uint64
// Answered is every request that got a reply, including one that reported
// [ErrResponderFailed].
Answered uint64
// NoResponders is every request sent to a subject nothing subscribed to.
NoResponders uint64
// NoReply is every request whose context ended before a reply.
NoReply uint64
// Failed is every request the transport refused, or that Stop cancelled.
Failed uint64
// Unclaimed is every reply that arrived after its request had finished.
// Only the total from [Bus.RequestStats] carries it.
Unclaimed uint64
}
RequestStats is the asking side's accounting, cumulative for the life of the bus (0011 D7). Once nothing is in flight, Sent equals Answered + NoResponders + NoReply + Failed. Unclaimed counts replies, not requests, so it sits outside that identity.
type Responder ¶ added in v0.3.0
Responder answers one request. The returned event is the reply, so every invocation that completes answers exactly once (0011 D1).
An answer the requester did not want — a refusal, a conflict — is still a reply, in its payload. An error, a panic, or a reply that cannot be sent becomes a failure reply instead: the requester gets ErrResponderFailed at once rather than waiting out its deadline, and the error's text stays here, logged and counted. A delivery that was published rather than requested still runs the responder; its reply is discarded and counted Unaddressed.
type RestartPolicy ¶
type RestartPolicy = controls.RestartPolicy
RestartPolicy governs how a failing subscription is restarted.
It is go/controls' type rather than one of our own. The fields are already the right ones, and two policy types with identical fields is the kind of divergence nobody notices until they disagree.
type Session ¶
type Session interface {
// Publish sends one event to one subject. The subject is concrete: this
// module has already refused a wildcard.
//
// It must be safe to call concurrently with Source.Retire and with Close,
// and it must fail promptly once Close has returned. That is what makes a
// bounded stop honest: a publish that outlives the drain budget lands on a
// closing session and gets an error rather than causing corruption.
Publish(ctx context.Context, subject string, event cloudevents.Event) error
// Subscribe creates one source. The pattern has already been validated.
Subscribe(ctx context.Context, spec SourceSpec) (Source, error)
// Close releases everything this session acquired, including every source
// it created. It is idempotent, and a nil error means the resources are
// gone — which is the only thing that permits a new session to open.
Close(ctx context.Context) error
}
Session is one connected lifetime of a backend.
It exists so that sequential sessions are structural rather than a sentence in a contract: a bus generation holds a session, and a new generation cannot begin until the previous session has confirmed closure, so an adapter cannot be asked to Start something it never Stopped.
type Settings ¶
type Settings struct {
// Logger receives what the bus has to say. Nil discards.
Logger *slog.Logger
// Source identifies the context events from this bus happened in: a
// URI-reference naming the service, not the machine.
//
// Required by [Bus.PublishValue] and by nothing else. CloudEvents makes it
// half of an event's identity — "ID, with Source, uniquely identifies the
// event" — so a generated event without one is invalid, and generating an
// invalid event quietly is worse than refusing to.
//
// [Bus.Publish] does not need it, because there the caller builds the whole
// event and owns that field.
Source string
// Codec turns a value into a payload for PublishValue and back for
// Delivery.Decode. Nil means [JSONCodec], which is what this module already
// assumed, so an existing caller sees no change.
//
// One codec per bus, deliberately. A bus with two is two buses wearing one
// name, and a consumer reading a subscription cannot tell which applies
// without looking elsewhere. If a real case appears for mixing formats,
// per-subscription is an additive change later; the reverse is not.
Codec Codec
// Subscriptions is the set known at wiring, and the set Ready reports on
// (0001 D5).
//
// It may be empty, for a process that only publishes. It is not the whole
// set for the bus's life: [Bus.Subscribe] attaches one afterwards, for a
// set not knowable at construction or to take a bigger share of a burst. Declaring one here is what a consumer does when it can,
// because a wiring-time subscription is running before Ready closes and
// needs no error path of its own.
Subscriptions []SubscriptionSpec
// RequestNames are the names [WithRequestName] may file a request's
// counters under. Declared here so the number of rows is fixed by the code,
// as subscription names are (0011 D8).
RequestNames []string
}
Settings configure a bus.
type Settlement ¶ added in v0.5.0
type Settlement struct {
Outcome Outcome
// Delay is how long a declined delivery, Nacked or Silent, must wait
// before its next attempt (0013 D1). Zero is at once. Only ever set for a
// backend that declares Capabilities.DelaysRedelivery.
Delay time.Duration
}
Settlement is what the bus asks a backend to do with one delivery.
A struct rather than more parameters on the callback, because this seam has grown before and will again (0013 D4).
type SettlementCounters ¶
type SettlementCounters struct {
Acked uint64
Nacked uint64
Silent uint64
// Refused is settlements the backend could not perform, whatever the
// handler said. Every refusal leaves the message unacknowledged at the
// backend.
Refused uint64
}
SettlementCounters is what handlers said, per attempt that reached one, and separately whether the backend could act on it.
Two axes, not four categories. Acked, Nacked and Silent are HANDLER INTENT and partition Delivered. Refused is BACKEND OUTCOME and is orthogonal: a nacked attempt whose nack could not be delivered moves Nacked AND Refused.
func (SettlementCounters) Total ¶
func (s SettlementCounters) Total() uint64
Total is every settled attempt, whatever the handler said. It excludes Refused, which is on the other axis.
type Shape ¶
type Shape string
Shape is how a message reaches the subscribers of a subject.
Both are core rather than capabilities, because a spike measured them portable across three runs on both backends: fan-out delivered to every subscriber and a group delivered to exactly one member, every time. What is NOT portable is behaviour when a member is degraded; see Capabilities.
const ( // Fanout delivers a copy to every subscriber of the subject. Fanout Shape = "fanout" // Competing delivers to exactly one member of a named group. // // Exactly one, and no more than that is promised. Which member, whether the // split is even, and whether a degraded member is routed around are all // backend behaviour: NATS distributes rather than load-balances, so a wedged // member keeps receiving its share, fills, and sheds while a healthy member // idles. Competing Shape = "competing" )
type SlowHandlerPolicy ¶
type SlowHandlerPolicy uint8
SlowHandlerPolicy is what a backend does with a message whose handler is still running when AckWait expires (0005 D7). A broker cannot tell slow from dead, and the overlap spike measured one message running in three handlers at once when it guessed wrong.
const ( // ExtendLease keeps the message this handler's, for up to // MaxProcessingTime. Handlers must be IDEMPOTENT: this does not eliminate // concurrent execution and nothing can, because a partition or a failover // still hands the message on. The zero value, because Redeliver's failure // is silent duplicate work under load. ExtendLease SlowHandlerPolicy = iota // Redeliver lets the backend hand the message on while this handler may // still be running. Handlers must be safe under CONCURRENT EXECUTION of // the same message, which usually means a lock or a unique constraint. Redeliver )
func (SlowHandlerPolicy) String ¶
func (p SlowHandlerPolicy) String() string
type Source ¶
type Source interface {
// Deliveries is the boundary this module owns (0001 D12). Subscriber code
// never runs on a backend goroutine; it consumes from here.
//
// # The channel's capacity is the backend's decision, and it is load-bearing
//
// For a backend whose queue lives elsewhere, a NATS subscription's pending
// buffer, this channel is UNBUFFERED: the adapter's handler blocks until
// this module takes the message, which applies backpressure into the queue
// the backend already owns and sheds under the backend's own policy. A
// buffered channel there would be a second queue in front of the first,
// which is the arrangement 0001 D14 refuses.
//
// For a backend whose queue IS a channel, the in-memory one, this channel is
// that queue and its capacity is the bound.
//
// It is closed by Retire, so a consumer ranging over it terminates.
Deliveries() <-chan Delivery
// Snapshot is a best-effort, constant-time reading of this source's
// counters, for observability. On a NATS backend it is a local read of the
// subscription's pending count, not a round trip, so watching it costs
// nothing.
//
// # It is advisory, and Retire is not
//
// A backend that owns its counters may make this coherent, and the
// in-memory one does. A backend whose counters live inside somebody else's
// object may not: NATS reports Delivered and Pending through two separate
// calls that nothing here can make atomic with respect to each other or to
// the traffic between them. Demanding coherence of every backend would be
// asserting an atomicity a transport cannot give, and the conformance suite
// could not tell the difference.
//
// So accounting that must be exact reads Retire's SourceFinal, which is
// stable because nothing moves after a source has retired. Valid until
// Retire returns; Retire's own result is the final reading.
Snapshot() SourceStats
// Retire atomically closes admission, removes this source from any routing
// or load-balancing choice, waits for enqueues already in progress, closes
// Deliveries, and returns counters that will never change again.
//
// # Why one call rather than Close plus a counter read
//
// Separate operations cannot express this. Between a final Depth read and a
// Close, an enqueue can land and be lost from the accounting; and a NATS
// subscription reports Dropped() as -1 once invalidated (0001 D17), so the
// last chance to read a true count is before the close that would have made
// it unreadable. Retiring and reporting in one step removes the window.
//
// It is idempotent: repeated calls return the same final result.
//
// # An error does not assert cleanup
//
// A non-nil error means the source may still be live, so the core retries
// rather than proceeding. Success is the only permission to reactivate,
// because a source abandoned while still routed becomes a ghost member of
// its competing group, which is worse than being unavailable.
Retire(ctx context.Context) (SourceFinal, error)
}
Source is one generation of a live subscription, owned by the backend.
It is called a source rather than a subscription because a subscription outlives the thing delivering to it: a consumer that restarts keeps its name, its ledger row and its place in a competing group, and gets a new source.
type SourceFinal ¶
type SourceFinal struct {
SourceStats
// Cause is why the source ended, or nil when the core retired it (0010 D3).
Cause error
}
SourceFinal is a retired source's last word. Its counters never change again, which is what makes it safe to fold into a cumulative ledger exactly once.
type SourceID ¶
type SourceID struct {
// Activation is the bus activation that owns this source. It increases by
// one per successful Start and never repeats.
Activation uint64
// Consumer is the subscription name, stable across activations.
Consumer string
// Attempt increases by one per source this consumer has created within one
// activation, so a restarting consumer's generations stay distinguishable.
Attempt uint64
}
SourceID identifies one source across the generations that may replace it.
It exists because a restart recreates a consumer's source (0002 D2), so a counter read from the source itself resets. Tagging every report with the generation that produced it is what lets the core aggregate across a restart instead of silently losing every discard from every prior generation.
type SourceSpec ¶
type SourceSpec struct {
// ID identifies this source and the generation that owns it.
ID SourceID
// Name is the subscription name, for error messages and attribution.
Name string
// Pattern has already been validated by this module.
Pattern string
// Shape is fan-out or competing.
Shape Shape
// Group is the competing-consumer group name, empty for fan-out.
Group string
// Overflow is the policy this source must apply at its bound.
Overflow Overflow
// Mode is the delivery mode this source must honour.
Mode DeliveryMode
// MaxDeliver bounds redelivery under AtLeastOnce.
MaxDeliver int
// DurableName is the broker-side identity under AtLeastOnce, already
// validated. A backend that survives a restart resumes it; one that does
// not has nothing to resume.
DurableName string
// StartAt applies when DurableName is first created. Reopening it with a
// different value is the backend's to refuse, naming the field (0005 D10).
StartAt StartPosition
// SlowHandler, AckWait and MaxProcessingTime are the lease, which the
// backend owns (0005 D7): AckWait starts at broker delivery, before the
// core holds anything, so nothing on the core side could keep it.
// MaxProcessingTime arrives with the default already applied under
// ExtendLease, and zero under Redeliver. A backend with no lease — one
// that never hands a message on while a handler runs — has nothing to do
// with any of them.
SlowHandler SlowHandlerPolicy
AckWait time.Duration
MaxProcessingTime time.Duration
// Bound is how many messages this source may hold, with the default
// already applied: never zero, never negative. A backend applies it to
// whatever its queue is — a channel's capacity, a pending limit, a pull
// batch — and what happens when it is reached is the backend's to offer.
Bound int
}
SourceSpec is what a backend needs to create a source: the caller's subscription declaration, minus everything that is none of a backend's business.
It exists because SubscriptionSpec carries a Handler (0002, OQ4) and a restart policy, and handing those to a transport adapter would invite it to call one. The projection is built by this module, so a backend cannot see subscriber code at all.
type SourceStats ¶
type SourceStats struct {
ID SourceID
// Accepted is everything that entered this source's queue.
Accepted uint64
// Pending is accepted but not yet handed to the core.
Pending uint64
// Refused is everything turned away at the queue bound: a newcomer that
// never entered the queue. It IS an offer — the publisher handed it to this
// source — so it counts toward Offered, and it is a loss.
Refused uint64
// Evicted is everything removed from the queue to make room: an incumbent
// that had ALREADY been counted in Accepted. It is a loss and NOT a new
// offer.
//
// These were one field, Shed, and it did both jobs. Offered was derived as
// Accepted + Shed, so under ShedOldest an evicted message was counted on
// entry and again on eviction: measured at Offered 3743 against 2000
// published (issue 5). ShedNewest balanced only because a refused newcomer
// never entered Accepted. The seam could not tell the two apart, and now
// it says which.
Evicted uint64
// Exhausted counts MESSAGES that ran out of attempts under AtLeastOnce. A
// message, not an attempt, which is why it cannot live in the attempt
// identity (0006 D2, D8).
//
// This is the one field durable delivery adds to the seam. The vertical
// slice also added Released, and an architectural review showed every
// released case — panic, shed, retired backlog — is an event the core
// already sees, so the core classifies it once the ledger knows the
// subscription's mode. Exhausted genuinely cannot be derived: the backend
// owns MaxDeliver.
Exhausted uint64
// LeaseExpired counts ATTEMPTS whose lease reached MaxProcessingTime and
// were handed on while the handler may still have been running (0005 D8).
// The second field durable delivery adds to the seam, for the same reason
// as the first: the backend owns the lease, so only it can know. Attempt
// grain, because one message can reach the bound twice. Neither a loss nor
// a release — the attempt is still running and is Delivered when its
// handler returns — so it sits beside the identity, not inside it. A
// backend with no lease reports zero.
LeaseExpired uint64
// DelayShortened counts declines whose delay the backend could not hold in
// full, and redelivered sooner than asked (0013 D9). Beside the identity,
// like LeaseExpired: the attempt was still made. A backend with no limit
// on a delay reports zero.
DelayShortened uint64
}
SourceStats is one coherent reading of a source's counters.
The three numbers are related by an identity the conformance suite asserts: at any moment Accepted equals what has been handed to the core plus Pending, and Shed is everything refused at the bound. A backend that cannot report all three cannot be bounded-and-counted, which is this module's one promise.
func (SourceStats) Shed ¶
func (s SourceStats) Shed() uint64
Shed is everything lost at the queue bound, whichever way it went.
The old single field under its old name, as a method, so that a reader who only wants "was anything lost at the bound" keeps that question — and so that the ledger's loss total is Refused + Evicted rather than one of them.
type StartPosition ¶
type StartPosition uint8
StartPosition is where a NEW durable identity begins reading. It applies when the identity is created and never again: reopening an existing DurableName with a different value is the backend's to refuse, naming the field (0005 D10).
const ( // NewMessages begins at the moment of creation. The zero value. NewMessages StartPosition = iota // AllAvailable begins at the oldest message the backend still retains, // which on a backend that retains nothing is NewMessages by another name. AllAvailable )
func (StartPosition) String ¶
func (p StartPosition) String() string
type Status ¶
type Status struct {
Generation uint64
State LifecycleState
Cause error
// Consumers is a fresh copy, so a reader cannot race the writer.
Consumers map[string]ConsumerStatus
// Changed closes on the next status change.
Changed <-chan struct{}
}
Status is a coherent reading of the bus.
State and Changed are captured under one lock, which is what lets a caller wait for a transition without missing one: a bare generation number cannot express "wait until this differs" without a window in which Running became Stopping and became Running again.
type SubscriptionSpec ¶
type SubscriptionSpec struct {
// Name identifies the subscription in counters, health and supervision. It
// is this module's handle rather than the backend's, because a discard has
// to be attributed to a consumer and a pattern is not a consumer.
Name string
// Pattern is the subject pattern, validated by this module before a backend
// sees it (D18).
Pattern string
// Shape is Fanout or Competing.
Shape Shape
// Group names the competing group, and is required for Competing and
// refused otherwise.
Group string
// Overflow is the policy requested at the bound. OverflowNative asks for
// nothing.
Overflow Overflow
// Mode is how many times a message may be attempted. Zero is AtMostOnce,
// which is today's behaviour. A mode the backend does not declare is a
// construction error, never a silent downgrade (0005 D11).
Mode DeliveryMode
// MaxDeliver bounds redelivery under AtLeastOnce, and is required there.
// Refused under AtMostOnce, where it would mean nothing.
MaxDeliver int
// AllowVolatile lets an AtLeastOnce subscription run on a backend whose
// unacknowledged messages do NOT survive a restart.
//
// The zero value refuses. Asking for at-least-once and saying nothing about
// persistence gets the safe reading: a backend that would lose your
// backlog on a bounce is refused at construction, and you opt into
// volatile retry deliberately, by name (0005 D13). Refused under
// AtMostOnce, where it means nothing.
AllowVolatile bool
// DurableName is the identity a backend that survives a restart uses to
// recognise the same consumer after a bounce. Required under AtLeastOnce
// on EVERY backend, so a specification valid in a test against memory is
// valid in production against a broker; refused under AtMostOnce, which
// has no identity to keep. Distinct from Name, the local handle, because
// conflating the two is how a rename silently starts a new consumer from
// scratch (0005 D4). One per bus (D10). Under Competing it must equal
// Group: a competing group is one shared durable identity, and a name is
// one consumer server-wide — every process on it competes, whatever Shape
// it declared, so fan-out across replicas needs one name per replica.
DurableName string
// StartAt is where a NEW durable identity begins reading, applied at
// creation only (0005 D10). Refused under AtMostOnce.
StartAt StartPosition
// SlowHandler is what the backend does with a message whose handler is
// still running when AckWait expires: ExtendLease, the zero value, or
// Redeliver. Refused under AtMostOnce, which has no lease.
SlowHandler SlowHandlerPolicy
// AckWait is how long the backend waits for a settlement before deciding
// the consumer is gone. Under Redeliver it is a CORRECTNESS setting:
// shorter than the slowest handler, and concurrent duplicate work is
// routine. Under ExtendLease it governs only how fast a dead consumer is
// noticed. Zero leaves the backend's own default. Refused under
// AtMostOnce.
AckWait time.Duration
// MaxProcessingTime bounds ExtendLease; zero means
// DefaultMaxProcessingTime. Reaching it is reported rather than quietly
// applied: extension stops, AckWait expires, and the message is handed on
// while the handler may still be running. Refused under Redeliver, which
// has no lease to bound, and under AtMostOnce.
MaxProcessingTime time.Duration
// Backoff is how long a declined message waits before its next attempt,
// by attempt number. The zero value redelivers at once. Refused under
// AtMostOnce, and on a backend that does not declare DelaysRedelivery
// (0013 D3, D8).
Backoff Backoff
// MaxNackDelay is the longest delay Delivery.NackAfter may ask for; zero
// takes Backoff's Max. With neither set, NackAfter is refused. Refused
// under AtMostOnce, and when below Backoff's Max.
MaxNackDelay time.Duration
// Bound is how many messages this subscription may hold; zero means
// DefaultQueueBound. The caller's, because 256 is meaningless without the
// message size, and because it has to relate to a backend's unacked window:
// a window larger than the bound sheds messages the backend already handed
// over, which under AtLeastOnce are released and immediately redelivered
// (0001 D7, amended). Negative is refused: unbounded is not a policy.
Bound int
// Handler receives every delivery. A subscription has exactly one of
// Handler and Responder: neither, or both, is a construction error, not a
// runtime state.
//
// It lives here rather than being bound later (0002 OQ4) because a bus that
// polls for a handler which may never arrive cannot honestly say it
// installed everything a running bus needs. Binding it at construction
// makes an accidentally handlerless live subscription unrepresentable, and
// removes a name written in two places that has already produced a defect.
Handler Handler
// Responder answers requests instead of handling deliveries: its returned
// event is the reply (0011 D1). A responder is Competing and AtMostOnce,
// and needs Settings.Source to build a failure reply.
Responder Responder
// Concurrency is how many goroutines read this subscription's deliveries.
// Zero means one.
//
// At one, delivery is ordered. Above one it is concurrent and explicitly
// unordered, decided here at wiring rather than at the moment a pool
// happens to scale (0002 D3). The readers share the queue the backend
// already owns; this module adds none.
Concurrency int
// RestartPolicy governs how this subscription is restarted when its
// delivery loop fails, and nil means never restart.
//
// It is go/controls' type rather than one of our own, because the fields
// are already the right ones and two policy types with identical fields is
// the divergence nobody notices until they disagree. Two of its fields are
// inert on a subscription: HealthFailureThreshold and HealthCheckInterval
// drive a Service's health-based restarts through its Status probe, and a
// supervised child has no probe.
RestartPolicy *controls.RestartPolicy
}
SubscriptionSpec is a subscription as intent, before a backend has created it.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package conformance is the suite every messaging.Backend must pass.
|
Package conformance is the suite every messaging.Backend must pass. |
|
Package memory is a messaging backend built on channels, depending on nothing.
|
Package memory is a messaging backend built on channels, depending on nothing. |
|
Package test helps a consumer look for a handler panic before one happens.
|
Package test helps a consumer look for a handler panic before one happens. |