The backend contract¶

type Backend interface {
    // Capabilities is asked before anything is built.
    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.
    Open(ctx context.Context) (Session, error)
}

type Session interface {
    Publish(ctx context.Context, subject string, event cloudevents.Event) error
    Subscribe(ctx context.Context, spec SourceSpec) (Source, error)
    Close(ctx context.Context) error
}

The contract carries this module's semantics rather than the union of every broker's. A Backend value itself is cheap and stateless: its constructor, memory.New() for the in-memory one, acquires nothing, and everything a connected lifetime needs lives on the Session that Open returns. That is deliberate: it replaces an earlier Start/Stop pair on Backend itself precisely so that "what did starting this backend acquire" has an answer a caller can hold and close, and so that a bus generation cannot observe a previous session's leftovers.

What it refuses on your behalf, unless declared otherwise¶

• No persistence, no replay, by default. Nothing survives a restart and a subscriber that starts late has missed what came before, unless the backend declares SurvivesRestart and, for a new identity, Replays.
• No delivery guarantee stronger than AtMostOnce, unless declared. A backend adds AtLeastOnce to Capabilities.DeliveryModes to offer it, and a subscription that asks for a mode the backend has not declared is refused at construction rather than silently downgraded.
• No ordering across subscribers, ever. Each subscriber sees its own messages in order; nothing coordinates two with each other. Per-subscriber ordering itself is a declared capability (Capabilities.Ordered) and, on the core side, holds only at SubscriptionSpec.Concurrency: 1.

The guarantee under all of it, declared or not, is that a failure is counted, never silently absorbed.

What it requires of an implementer¶

Every subscriber gets its own copy of the mutable parts. A CloudEvent carries a byte slice and a map, so handing the same value to every fan-out subscriber shares mutable storage between them: one handler unmarshalling in place or annotating an extension corrupts what its siblings receive, and between two live handlers it is a data race as well.

messaging.CopyEvent does it, so no backend has to work out what "its own" means, and the conformance suite checks it.

Everything Open acquires must be reachable from the Session it returns, or it leaks. That is the one obligation this contract places on an implementer: a Session.Close that cannot reach something Open acquired cannot release it, and a new session cannot honestly be opened while the old one might still hold it.

Building a delivery¶

A backend hands the bus a Delivery through NewDelivery, never as a struct literal:

messaging.NewDelivery(messaging.DeliveryOptions{
    Subject:  subject,
    Event:    messaging.CopyEvent(event),
    Attempt:  n,        // the backend's delivery count for this message; zero reads as one
    OnSettle: settle,   // nil for a backend that offers no settlement
})

The settlement state behind Ack and Nack is unexported and shared across every copy of the value, so that a handler holding two copies cannot make them disagree. That is why a backend cannot build one directly. OnSettle is called exactly once, after the handler returns, with the outcome the handler declared — Silent if it said nothing — and returns an error if the backend could not act on it, which the bus counts as Settlement.Refused. The context is the reader's and is cancelled during a bounded Stop, so a settlement that misses the shutdown budget is refused rather than blocking it.

An adapter must treat an Outcome it does not recognise as Nacked. Outcome will grow — a terminal refusal for a poison message is the first candidate — and an adapter compiled against today's values will one day be handed a newer one. The safe reading of a value you do not know is "did not accept", which is the reading Silent already gets.

DeliveryOptions is a struct rather than positional parameters because this constructor is public API at a seam that grows, and an adapter in another module compiled against a positional signature breaks on every addition.

Proving you survive a restart¶

A backend declaring SurvivesRestart is held to the rows only a durable backend can be held to. None of them can be run by conformance.Run, whose New builds an independent backend per call; they need something that owns the storage. So such a backend supplies a conformance.Durable — Open(t) builds a backend over the fixture's storage, Restart(t) bounces the storage keeping its contents — and runs conformance.RunDurable, which is everything Run does plus:

Run refuses a backend that declares SurvivesRestart: a capability the suite cannot exercise is a promise nobody keeps. Every row above is proved in the suite's own tests both ways — green against the smallest store that genuinely has the property, red against the in-memory backend wrapped to claim it — so that a row guards something before any backend meets it.

Two rows read the fixture as well as the backend. A fixture may implement conformance.Immutable to name which of Pattern, StartAt, MaxDeliver and AckWait its backend refuses on reopen — a queue-shaped backend holds a queue, not the whole identity, and its pattern is the operator's binding — and conformance.Granular to state the smallest lease it can honour, so the lease row's timings scale rather than asking for what cannot exist. A fixture that implements neither is held to all four fields at millisecond timings.

The restart-attempt row accepts what every transport can promise. What an attempt is differs: a receive on SQS, a send on JetStream, a failed settlement on AMQP, a handler returning in this contract. A redelivery fetched and never handled before the restart may already have been spent, so the row asserts that the count did not reset and the budget was not exceeded — the next attempt is numbered from the budget's remainder, or the message was exhausted instead — rather than one exact number.

A managed service's fixture restarts what it can. Durable.Restart is meant to bounce the storage; where the storage is a service the fixture cannot restart, it restarts every client and backend and says so, and what the rows prove there is that a consumer's restart loses no message and no attempt count. A backend declaring no overflow policy is held to a backpressure form of the bound row: under flood it holds at most the bound, sheds nothing, and delivers everything once drained.

messaging.Settle(ctx, d) performs the settlement a handler declared: it is what the bus does after a handler returns, exported so the conformance suite — which drives a backend at this seam with no bus above it — can complete the exchange the same way. A handler never calls it. It settles at most once per delivery; a second call is a no-op.

Source¶

type Source interface {
    Deliveries() <-chan Delivery
    Snapshot() SourceStats
    Retire(ctx context.Context) (SourceFinal, error)
}

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.

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, where it sheds under the backend's own policy. A buffered channel there would be a second queue in front of the first.

For a backend whose queue is a channel, this channel is that queue and its capacity is the bound.

Snapshot is a best-effort, constant-time reading for observability, and is advisory: a backend whose counters live inside somebody else's object, NATS reporting Delivered and Pending through two separate calls, cannot make it atomic with respect to the traffic between them. Accounting that must be exact reads Retire's result instead, which is stable because nothing moves after a source has retired.

Retire does what Close plus a separate counter read cannot: atomically close admission, remove the source from any routing or load-balancing choice, wait for enqueues already in progress, close Deliveries, and return counters that will never change again. It is idempotent, and a non-nil error means the source may still be live, so the core retries rather than treating an unsettled source as gone.

Capabilities¶

type Capabilities struct {
    Overflow                []Overflow
    Patterns                bool
    DeliveryModes           []DeliveryMode
    SurvivesRestart         bool
    ExtendsLease            bool
    Ordered                 bool
    Replays                 bool
    RestartLosesLastFailure bool
    RoutesAroundDegraded    bool
}

Declared before anything is built, rather than discovered by trying, because construction has to degrade honestly and cannot do that by attempting an operation and seeing what happens.

DeliveryModes lists every mode the backend offers, including AtMostOnce. An empty list offers nothing — SQS and Pub/Sub cannot promise never-duplicated delivery, so "every backend can do at-most-once" is false and the zero value is not implied. A subscription asking for a mode the backend did not declare is refused at construction, fatally, never downgraded: a delivery guarantee is this module's, and silently changing it changes the handler's obligations without telling the author.

SurvivesRestart is a storage property on its own axis: whether an unacknowledged message outlives the process. It is separate from the modes because a backend can retry until acknowledged and still lose its backlog on a bounce, and a subscription needs to be able to ask which. The in-memory backend offers AtLeastOnce and reports false — at-least-once within a process run, honestly — so a subscription gets that mode there only by setting AllowVolatile.

ExtendsLease says whether the backend can keep a message assigned to a handler that is still working. ExtendLease is the default SlowHandler policy, so the capability has to say what happens without it: a backend that survives a restart and cannot extend refuses an ExtendLease subscription at construction, naming SlowHandler and asking for Redeliver. A backend with no lease at all — the in-memory one never hands a message on while a handler runs — has nothing to refuse and declares false honestly. A lease exists only where a message can outlive the process holding it, which is what SurvivesRestart already says.

Ordered says 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 queue does, at the price of one message in flight per subject, which is why the two are one declaration. The suite holds an ordered backend to the ordering row, and runs the per-message-acknowledgement row across two subjects on it rather than assuming two in flight on one.

Replays says 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: AllAvailable on a backend declaring Replays: false is refused at construction, naming the subscription, because a queue delivers what it holds and "from the start" would be a promise nobody keeps. Likewise a wildcard pattern on a backend declaring Patterns: false is refused at construction rather than passed through for the backend to approximate.

SourceStats and SourceFinal carry two more fields for the mode, both numbers only the backend can know. Exhausted is messages that ran out of attempts: the backend owns MaxDeliver. LeaseExpired is attempts whose lease reached MaxProcessingTime and were handed on while the handler may still have been running: the backend owns the lease. Everything else an at-least-once backend abandons — a panic, a shed attempt, a retired backlog — the core classifies itself once it knows the mode. A backend with no lease reports LeaseExpired as zero and is not held to a row for it.

RestartLosesLastFailure says 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 normally 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), so 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 silent lie of SurvivesRestart. Meaningless without SurvivesRestart, and false on every backend that persists a failure when it happens.

RoutesAroundDegraded is there because it is the difference a consumer is most likely to assume the wrong way round. "Queue group" suggests load balancing; NATS distributes and does not load balance, so a wedged member keeps receiving its share, fills, and sheds while a healthy member sits idle. Measured: 11, 13 and 14 messages lost across three runs of 40, with the healthy member dropping none. An in-memory backend routing least-loaded loses none of it.

SubscriptionSpec¶

Under AtLeastOnce, ShedNewest and ShedOldest are refused: a message shed at the bound was never acknowledged, so the backend owes it again, and on a volatile backend "owed again" means gone. The bound under that mode means the source stops fetching; what remains is Fail.

A DurableName must be one a broker can hold: no whitespace, dots, wildcards, path separators or non-printable characters. The bus refuses the rest before a backend sees it, so the refusal is the same on every backend and arrives at construction rather than at the broker. The SourceSpec a backend receives carries DurableName and StartAt as declared; a backend that survives a restart resumes the identity, refuses reopening it with a different Pattern, StartAt, MaxDeliver or mode by naming the field, and applies StartAt only when it creates the identity.

The lease is the backend's. AckWait starts at broker delivery, while the message is parked on the unbuffered channel send and before any handler runs, so nothing on the core side could keep it alive. The SourceSpec carries SlowHandler, AckWait and MaxProcessingTime, the last with the default already applied under ExtendLease and zero under Redeliver. A backend that offers ExtendLease heartbeats from receipt until its settlement reply fires or MaxProcessingTime is reached, and reaching it must be visible rather than quietly applied. A backend with no lease at all — the in-memory one never hands a message on while a handler is running — has nothing to do with any of the three, and says so by leaving them unread.

Two RestartPolicy 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.

Overflow¶

Core NATS can offer only ShedNewest: its behaviour is fixed, with no hook to evict the oldest and no way to block a publisher that is already fire-and-forget into a write buffer. The in-memory backend offers three, ShedNewest, ShedOldest and Fail, because a channel it owns can implement any of them; Block is defined but currently declared by no backend in this module, having been removed from the in-memory backend once it was the reason a bounded Stop was hard to prove correct.

Swapping the backend can therefore change what happens at a full queue, and a caller who cares must ask. That is the honest position rather than a comfortable one.