Proposal: RabbitMQ

[Zaba505]

Note

Part of the humus re-evaluation; see Proposal: humus on the bedrock process model, wave 4. It builds on Proposal: queue processing and the Kafka runtime, whose queue contract it implements, on Proposal: humus.Main, and on the wave-1 proposals: the lifecycle, configuration, OpenTelemetry without globals and humustest. It depends on no other wave-4 proposal. It blocks Proposal: failure injection (wave 5), which strips scenarios at this Service and adds an injection point to its Publisher, and Proposal: walkthroughs and a humus MIGRATING.md (wave 6), whose RabbitMQ processor (#318) is built on it.

Summary

• queue/rabbitmq is a new nested module. Its Service consumes one RabbitMQ queue with amqp091-go, RabbitMQ's own Go client, and hands each delivery to a queue.Processor[rabbitmq.Message]. It implements queue's contract exactly as queue/kafka does, and shares no code with it.
• Delivery maps onto acknowledgements. Under AtLeastOnce a delivery is acked once Process returns nil, and a processing error rejects it and ends the Service, so the App exits 1 and the broker delivers it again. Under AtMostOnce a batch of deliveries is acked, and the ack confirmed, before any of it is processed. Neither uses auto-ack, so the prefetch count always bounds what the client holds.
• Shutdown drains, and counts nothing against a message. On Serve's context the Service cancels its consumer, lets the deliveries in flight finish and acks them, and returns everything it received but didn't process with one basic.nack, which RabbitMQ doesn't count toward a quorum queue's delivery limit.
• The broker is dialled through the Host. The Service dials s.Host.Dial, runs the TLS handshake itself and opens AMQP on that connection with amqp091's Open, so no URL, no net.Dial and no os.Args are involved. When the connection drops, the Service reconnects with bedrock's retry, as franz-go keeps retrying for Kafka.
• Consuming is a readiness monitor. It is false until the Service first consumes its queue, so a pod with a wrong broker address never turns ready, and stays true after that.
• A Publisher publishes with confirms. It is a client built in Init, with its own connection, publisher confirms and mandatory routing, bedrock's retry and circuit, and a span and two instruments per publish, following the conventions the charter sets for the wave-4 clients.
• Tests run on a fake broker over humustest's Host, inside synctest. No in-memory RabbitMQ exists, so humus writes one, rabbitmqtest. A prototype of the Service and a consume-only fake pass whole-App tests under -race. Container tests run against RabbitMQ in a CI job of their own, as for Kafka.

func main() {
	humus.Main(app.Init, humus.WithTelemetry(buildTelemetry))
}

app.Init sources the rabbitmq.Settings, builds the Processor, and returns the Service that rabbitmq.NewService(s, settings, p) builds.

Background

• humus has never had RabbitMQ. story(queue): implement a runtime for rabbitmq #310 asked for it before the reset, and the reset closed it. The charter's revision 7 put it in wave 4, on the queue contract: "It decides whether it also publishes."

• What wave 3 settled. queue is the contract between the end user's code and any broker (queue/doc.go):

  • a processing error under AtLeastOnce ends the Service, and the message is delivered again when the App restarts;
  • a processing error under AtMostOnce is logged and counted, and processing goes on;
  • no panic is recovered;
  • shutdown drains, and Process's context is cancelled only when the grace period runs out;
  • retries are the Processor's.

  The Kafka proposal sketched how RabbitMQ would fit: ack after Process, stop on an error so the broker requeues, and auto-ack for AtMostOnce. This proposal keeps the first two and replaces auto-ack (see Design decisions).

• The shape queue/kafka set. A nested module, a Service built by NewService(s, settings, p) that opens no connection, plain Settings with a check per field, telemetry from s.Telemetry with semconv's messaging conventions, and no App, no Main and no defaults of its own, because humus.Main's 20s grace period and no drain fit a consumer (its revision). This proposal follows it wherever RabbitMQ allows.

• A lesson from Kafka. queue/kafka: a fenced-out member's late commit can move the new owner's committed offset #757 found that a fenced-out member's late commit can move the new owner's offset. RabbitMQ's acks can't do that: a delivery tag belongs to one channel, and an ack sent on a channel that has closed fails in the client and never reaches the broker.

• amqp091-go v1.15.0 is the client: maintained by the RabbitMQ team, pure Go, and with one requirement, goleak, for its own tests (go.mod). Five of its properties shape this design:

  1. amqp.Open runs AMQP over a connection the caller has already opened. DialConfig takes a URL, and its Config.Dial has no context.
  2. It doesn't reconnect. v1.15.0 added Config.Recovery, marked experimental, which works only through DialConfig's URL.
  3. A goroutine buffers every delivery the broker sends, without a bound (consumers.go#L74). The prefetch count is the only bound, and "the server ignores this option when consumers are started with noAck" (channel.go#L912-L914).
  4. Given no consumer tag, it builds one from os.Args[0] (consumers.go#L19-L21). Its logger is a package-level variable, amqp091.Logger, a no-op by default (log.go).
  5. Its heartbeater and reader are goroutines on the connection it is given, with a ticker and read deadlines, so on humustest's net.Pipe network they run inside a synctest bubble.

• Quorum queues count failed deliveries. Since RabbitMQ 4.0 a quorum queue has a delivery limit of 20 by default. A message delivered more often than that is dead-lettered, if the queue has a dead-letter exchange, or dropped. basic.reject and a channel or connection that closes with the message unacked count toward the limit. basic.nack with requeue doesn't (quorum queues, Poison Message Handling).

Proposed API

package rabbitmq // github.com/z5labs/humus/queue/rabbitmq

// Connection is how a Service or a Publisher reaches the broker. Every field
// is a plain value, and a field's zero value means zero.
type Connection struct {
	// Addr is the broker's address, host:port. It is required.
	// Default: none. Valid: host:port, port 1 to 65535. Check: CheckAddr.
	Addr string

	// Vhost is the virtual host to open.
	// Default: "/". Valid: not empty. Check: CheckVhost.
	Vhost string

	// TLS is the config the broker is dialled with. nil means plaintext.
	// ServerName defaults to Addr's host. Default: nil. Check: none.
	TLS *tls.Config

	// SASL is how the connection authenticates: amqp091's PlainAuth,
	// AMQPlainAuth, or ExternalAuth for a client certificate in TLS. AMQP
	// always authenticates, so it is required. It is never changed.
	// Default: none. Valid: not nil. Check: CheckSASL.
	SASL amqp.Authentication

	// Heartbeat is the heartbeat interval the client proposes. The broker's,
	// if shorter, wins. A connection silent for 1.5 intervals is dead.
	// Default: 10s. Valid: whole seconds, 1s to 65535s. Check: CheckHeartbeat.
	Heartbeat time.Duration

	// ConnectTimeout bounds each dial, its TLS handshake and its AMQP
	// handshake. Default: 30s. Valid: positive. Check: CheckConnectTimeout.
	ConnectTimeout time.Duration
}

// Settings configures a Service.
type Settings struct {
	Connection Connection

	// Queue is the queue the Service consumes. It must exist: the Service
	// declares nothing. Default: none. Valid: 1 to 255 bytes. Check: CheckQueue.
	Queue string

	// Delivery is when a delivery is acked, relative to processing it.
	// Default: queue.AtLeastOnce. Check: queue.CheckDelivery.
	Delivery queue.Delivery

	// Prefetch bounds the deliveries the broker sends and the Service holds
	// unacked. Under AtMostOnce it also bounds the batch acked at once.
	// Default: 10. Valid: 1 to 65535. Check: CheckPrefetch.
	Prefetch int

	// Concurrency is how many deliveries are processed at once. At 1 they
	// are processed one at a time, in the order the broker delivered them.
	// Default: 1. Valid: 1 to Prefetch. Check: CheckConcurrency.
	Concurrency int
}

func DefaultSettings() Settings

func CheckAddr(addr string) error
func CheckVhost(vhost string) error
func CheckSASL(a amqp.Authentication) error
func CheckHeartbeat(d time.Duration) error
func CheckConnectTimeout(d time.Duration) error
func CheckQueue(name string) error
func CheckPrefetch(n int) error
func CheckConcurrency(n int) error

// Message is a delivery as a Processor sees it.
type Message struct {
	Queue       string // the queue it was consumed from
	Exchange    string // "" for the default exchange
	RoutingKey  string
	Redelivered bool           // the broker delivered it before
	Headers     map[string]any // field-table values, as amqp091 decodes them

	ContentType     string
	ContentEncoding string
	Persistent      bool
	Priority        uint8
	CorrelationID   string
	ReplyTo         string
	Expiration      string
	MessageID       string
	Timestamp       time.Time
	Type            string
	UserID          string
	AppID           string

	Body []byte
}

// Service consumes Settings.Queue and hands each delivery to a Processor.
type Service struct{ /* unexported */ }

// NewService checks settings, reporting every bad field in one error that
// names each one, and rejects a nil Processor. It opens no connection and
// registers no OnStop function: Serve connects.
func NewService(s *humus.Setup, settings Settings, p queue.Processor[Message]) (*Service, error)

func (svc *Service) Serve(ctx context.Context) error
func (svc *Service) Shutdown(ctx context.Context) error

// Consuming reports whether the Service has started consuming its queue:
// false until the broker first answers its basic.consume, and true from then
// until Serve returns. A lost connection after that doesn't make it false.
// It never returns an error, and its signature is health.MonitorFunc's:
//
//	health.Readiness(health.MonitorFunc(consumer.Consuming))
func (svc *Service) Consuming(ctx context.Context) (bool, error)

// Publishing is a message to publish. Its zero value is persistent.
type Publishing struct {
	Exchange   string // "" for the default exchange
	RoutingKey string
	Headers    map[string]any

	ContentType     string
	ContentEncoding string
	Transient       bool // not written to disk by the broker
	Priority        uint8
	CorrelationID   string
	ReplyTo         string
	Expiration      string
	MessageID       string
	Timestamp       time.Time
	Type            string
	AppID           string

	Body []byte
}

// ErrUnroutable is matched by the error of a publish the broker returned
// because no queue was bound to receive it.
var ErrUnroutable = errors.New("rabbitmq: message unroutable")

// PublisherSettings configures a Publisher. Its retry and breaker fields have
// the names, defaults and checks of the other wave-4 clients' settings.
type PublisherSettings struct {
	Connection Connection

	// Channels bounds the publishes in flight at once.
	// Default: 8. Valid: 1 to 65535. Check: CheckChannels.
	Channels int

	// Timeout bounds a whole Publish: every attempt, the waits between them,
	// and each confirm. A deadline the caller's context has already, if it
	// is earlier, wins. Default: 10s. Valid: positive. Check: CheckTimeout.
	Timeout time.Duration

	// MaxAttempts bounds the attempts of one Publish. 1 turns retries off.
	// Default: 3. Valid: 1 or more. Check: CheckMaxAttempts.
	MaxAttempts int

	// BackoffBase and BackoffMax are retry.Exponential's arguments.
	// Default: 100ms and 5s. Valid: positive, and BackoffMax at least
	// BackoffBase, which NewPublisher checks. Check: CheckBackoffBase,
	// CheckBackoffMax.
	BackoffBase time.Duration
	BackoffMax  time.Duration

	// RetryBudget limits retries across every publish.
	// Default: {MaxTokens: 10, TokenRatio: 0.1}. Check:
	// CheckRetryBudgetMaxTokens, CheckRetryBudgetTokenRatio.
	RetryBudget retry.BudgetConfig

	// Breaker configures the circuit breaker in front of the broker.
	// Default: {Window: 20, Probes: 1, OpenFor: 10s, ProbeTimeout: 10s}.
	// Check: CheckBreakerWindow, CheckBreakerProbes, CheckBreakerOpenFor,
	// CheckBreakerProbeTimeout.
	Breaker circuit.Config

	// BreakerPolicy decides when the breaker opens and closes. It is set in
	// code, not sourced. Default: open after 5 failures in a row, or when
	// half of the last 20 or more results failed; while half-open, reopen
	// on a failed probe and close after 3 successful ones. Valid: Trip and
	// Probe set, which NewPublisher checks. Check: none.
	BreakerPolicy circuit.Policy
}

func DefaultPublisherSettings() PublisherSettings

func CheckChannels(n int) error
func CheckTimeout(d time.Duration) error
func CheckMaxAttempts(n int) error
func CheckBackoffBase(d time.Duration) error
func CheckBackoffMax(d time.Duration) error
func CheckRetryBudgetMaxTokens(n int) error
func CheckRetryBudgetTokenRatio(r float64) error
func CheckBreakerWindow(n int) error
func CheckBreakerProbes(n int) error
func CheckBreakerOpenFor(d time.Duration) error
func CheckBreakerProbeTimeout(d time.Duration) error

// Publisher publishes to the broker over a connection of its own.
type Publisher struct{ /* unexported */ }

// NewPublisher checks settings and returns a Publisher. It opens no
// connection: the first Publish does. It registers the connection's close
// with s.OnStop.
func NewPublisher(s *humus.Setup, settings PublisherSettings) (*Publisher, error)

// Publish publishes m and returns once the broker has confirmed it. It is
// safe for concurrent use.
func (p *Publisher) Publish(ctx context.Context, m Publishing) error

amqp is github.com/rabbitmq/amqp091-go. Settings names its types in two places: SASL, as queue/kafka's names franz-go's sasl.Mechanism, and Message.Headers, whose nested tables and decimals are amqp091's.

Connecting

The Service and the Publisher connect the same way, with unexported code they share:

1. Dial Addr with s.Host.Dial, under a context bounded by ConnectTimeout.
2. With TLS set, run the handshake over that connection with tls.Client(...).HandshakeContext, on a clone of the config whose ServerName is Addr's host when the config leaves it empty. queue/kafka's dialer does the same.
3. Set the connection's deadline to the end of ConnectTimeout, and call amqp.Open with SASL, Vhost, Heartbeat and the locale. Open takes no context, so the deadline bounds the AMQP handshake. amqp091 clears it once the connection is open.

So humus never calls DialConfig: no URL, which would carry the password and be logged by mistake, and no DefaultDial, which calls net.DialTimeout. An end user who keeps an AMQP URI parses it with amqp.ParseURI in their Init and fills Connection from it.

Settings

Field	Default	Valid	Check
Connection.Addr	none	host:port, port 1 to 65535	CheckAddr
Connection.Vhost	"/"	not empty	CheckVhost
Connection.TLS	nil, plaintext	any
Connection.SASL	none	not nil	CheckSASL
Connection.Heartbeat	10s	whole seconds, 1s to 65535s	CheckHeartbeat
Connection.ConnectTimeout	30s	positive	CheckConnectTimeout
Queue	none	1 to 255 bytes	CheckQueue
Delivery	queue.AtLeastOnce	at-least-once, at-most-once	queue.CheckDelivery
Prefetch	10	1 to 65535	CheckPrefetch
Concurrency	1	1 to Prefetch	CheckConcurrency, and NewService for the bound

• Addr, SASL and Queue are required. humus imposes no port, as for httpserver and grpcserver, and AMQP has no anonymous login.
• Heartbeat and ConnectTimeout keep amqp091's defaults, 10s and 30s, which DialConfig applies. Heartbeats can't be turned off: a dead connection would then go unnoticed until TCP gave up.
• Prefetch of 10 and Concurrency of 1 keep a queue's order by default, hide the round trip of each ack behind nine more deliveries, and hold back at most ten messages from the queue's other consumers. Concurrency above Prefetch would leave workers that never get a delivery, so NewService rejects it, naming both fields.
• Delivery is the end user's design decision, as for Kafka, so most Inits set it in code, and DefaultSettings gives the safe one.

What Serve does

1. Connects, as above, and opens one channel. It sets basic.qos to Prefetch and consumes Queue with manual acks and a consumer tag of its own, so amqp091 never builds one from os.Args[0].
2. Hands deliveries to Concurrency workers. Each processes one at a time, with a context that keeps the values of Serve's context, carries the delivery's span, and is cancelled only when Shutdown's context is done.
3. Settles each delivery as its Delivery says (next section).
4. On shutdown, drains (see Shutdown). On a lost connection, reconnects (see Connection loss).

Delivery

AtLeastOnce. A worker acks its delivery once Process returns nil. When Process returns an error:

• the delivery is rejected with requeue, basic.reject, which counts toward a quorum queue's delivery limit;
• the Service stops as it does on shutdown: the other workers finish their deliveries and ack them, and everything received but not processed is nacked with requeue, which doesn't count;
• Serve returns the error, naming the queue and the message's ID when it has one, and the App exits 1.

So a message that always fails crash-loops the pod until the queue's delivery limit dead-letters or drops it: 20 deliveries on a quorum queue, by default since RabbitMQ 4.0. A classic queue has no delivery limit, and the crash loop is what shows the failure, as it does for Kafka. Nothing else counts toward the limit, so a message that was merely prefetched when a sibling failed is never dropped for it.

AtMostOnce. The Service takes a delivery and every delivery already buffered behind it, up to Prefetch, and acks them with one basic.ack with multiple set. Then it waits for the reply to a basic.qos on the same channel, and only then hands the batch to the workers. RabbitMQ handles a channel's methods in order, so the reply means the ack was processed. An AMQP ack has no reply of its own. If the connection drops before the reply arrives, the batch isn't processed: the broker either had the ack, and the batch is lost, or requeues it for someone else. Either way, nothing is processed twice.

This is queue/kafka's at-most-once, which commits a batch before processing any of it, with Prefetch in place of MaxPollRecords. A crash loses at most one batch, and a clean shutdown loses nothing: the Service processes the whole batch it acked, and nacks the rest. A processing error is logged at ERROR and counted, and processing goes on.

Order and concurrency

A queue delivers its messages to one consumer in order, so with Concurrency 1 the Service processes them in the order they were enqueued. Three things break that order, as with any RabbitMQ consumer: a redelivery, more than one replica, and Concurrency above 1. A Processor that needs per-key order uses a single-active-consumer queue, which is topology (see open question 2), and Concurrency 1.

Shutdown

When Serve's context is done, or Shutdown is called, the Service:

1. cancels its consumer, with basic.cancel, so the broker sends nothing more;
2. hands out no more deliveries, except the rest of an AtMostOnce batch it already acked;
3. waits for the workers' deliveries in flight to finish, and acks them;
4. nacks with requeue, in one basic.nack with multiple set, every delivery it received but didn't process;
5. closes the channel and the connection, and returns nil.

Nothing load-balances a consumer, so there is nothing to drain for: it stops taking work at once, as queue/kafka does. When Shutdown's context is done first, Shutdown cancels the context the deliveries in flight are processed with, closes the connection at once, waits for Serve, and returns an error wrapping its context's error, which the lifecycle reports as a *humus.GracePeriodError. The broker then requeues every unacked delivery, and that counts toward a quorum queue's delivery limit, because the grace period ran out.

Connection loss

A connection drops when the network does, when 1.5 heartbeat intervals pass with nothing read, or when the broker closes it. A node that shuts down for an upgrade closes every connection with 320 CONNECTION_FORCED. The broker requeues every delivery that was unacked on it. The Service then:

• lets the deliveries in flight finish. Their acks fail in the client, and the Service drops those errors: the broker already requeued the deliveries, and will deliver them again;
• logs the lost connection at WARN, with the AMQP code and reason when the broker gave one;
• reconnects with bedrock's retry: Exponential(100ms, 30s), no Budget, and math.MaxInt attempts, so that only Serve's context ends it. That is how retry's docs describe retrying a connection at startup. Each failed attempt is logged at WARN. The first connection is made the same way, so a broker that is down at startup is waited for, as franz-go waits for Kafka's.

Some errors aren't fixed by reconnecting. The Service stops, and Serve returns the error, when the broker:

• refuses the login (ACCESS_REFUSED, or amqp091's ErrCredentials or ErrSASL) or the vhost (ErrVhost);
• refuses the queue: 404 NOT_FOUND, 403 ACCESS_REFUSED or 405 RESOURCE_LOCKED, for a queue another connection consumes exclusively;
• closes the channel with 406 PRECONDITION_FAILED, such as a delivery held past the broker's consumer_timeout.

The broker cancelling the consumer, because its queue was deleted, is handled as a lost connection. The reconnect then finds the queue missing, and the Service stops.

Failures

	AtLeastOnce	AtMostOnce
Process returns an error	rejected, which counts toward the delivery limit; the Service stops as on shutdown, and Serve returns the error; exit 1	already acked; logged at ERROR and counted; processing goes on
Process panics	the process crashes; the broker requeues every unacked delivery, and each one counts	the process crashes; the batch acked is lost
the connection drops	in-flight deliveries finish, their acks fail, and the broker delivers them again; the Service reconnects	the same; a batch whose ack was confirmed is processed
the broker is unreachable	the Service keeps retrying, logging each attempt at WARN	the same
the login, vhost or queue is refused	the Service stops, and Serve returns the error; exit 1	the same

Telemetry

Everything comes from s.Telemetry, under the scope github.com/z5labs/humus/queue/rabbitmq, and follows semconv v1.43.0's RabbitMQ conventions, which humus pins as queue/kafka does.

• A span per delivery, named process <queue>, of kind CONSUMER, whose context Process is given. Its parent is what s.Telemetry.Propagator extracts from the message's headers, through an unexported carrier over Message.Headers that reads string values. A message whose headers carry no trace context starts a new trace. Its attributes are:

  • messaging.system (rabbitmq), and messaging.operation.type and messaging.operation.name, both process;
  • messaging.destination.name, which semconv defines as {exchange}:{routing key}:{queue}, leaving out what is empty and the routing key when it equals the queue. A message published to orders on the default exchange is orders;
  • messaging.rabbitmq.destination.routing_key when it isn't empty, and messaging.rabbitmq.message.delivery_tag;
  • messaging.message.id and messaging.message.conversation_id, the correlation ID, when they are set;
  • server.address and server.port, from Addr.

  A processing error is recorded on the span, with error.type and an error status. The body is never recorded.

• Two instruments from semconv's messagingconv: messaging.client.consumed.messages, counting the deliveries handed to the Processor, and messaging.process.duration, with error.type when Process fails. Each has messaging.system, messaging.operation.name, messaging.destination.name, server.address and server.port. An ack has no reply, so it has no duration to record. semconv's destination name includes the routing key, so a publisher that puts identifiers in routing keys gets a series per key.

• Logs. At INFO, consuming on each connection, with the queue and server.address. At WARN, each failed connection attempt and each lost connection. At ERROR, a processing error under AtMostOnce. An error that ends the Service is returned, and the lifecycle logs it.

• amqp091 logs nothing of its own here. Its logger is a package-level variable, which humus never sets, so its few lines stay on its no-op default. They are about its experimental recovery, which humus doesn't use, and two failures to answer the broker, which then surface as a lost connection. A forbidigo rule bans SetLogger in humus's code, beside OTel's globals.

The Publisher

The charter's three conventions for the wave-4 clients apply to it:

• It dials through Setup.Host and reports through Setup.Telemetry. NewPublisher opens nothing. The first Publish connects, as described under Connecting, on a connection of its own, as RabbitMQ advises, so that TCP back-pressure on publishing never holds up a consumer's acks. NewPublisher registers a function with s.OnStop that closes the connection within the grace period. The Publisher stays usable until then, so a Processor finishing its delivery during the drain can still publish.
• It composes retry and circuit as bedrock's docs do. Each Publish ranges over one Retrier's attempts, within Timeout, and each attempt goes through one Breaker, which stands in front of the broker. The Publisher builds one of each from its RetryBudget, Breaker and BreakerPolicy settings, which have the names, defaults and checks the HTTP and gRPC clients' settings have. Each transition is one log record, in the context of the publish that made it: circuit breaker open at WARN, and circuit breaker half-open and circuit breaker closed at INFO, with server.address and server.port from Addr, circuit.previous_state, and the counts of the state it left: circuit.successes, circuit.failures, circuit.consecutive_successes, circuit.consecutive_failures and circuit.timed_out. humus adds no metrics of its own for them.
• It names the seam failure injection uses: the attempt, inside the breaker and before the frame is written. An injected abort is then classified, counted and retried as a real failure would be. Wave 5 designs it.

An attempt takes a channel from a pool of Channels, connecting first if needed, and publishes m with mandatory set. Then it waits for the broker's confirm, or for ctx. Each channel is in confirm mode and carries one publish at a time. amqp091 hands a basic.return to the channel's listener before it handles the confirm that follows it, so a return that is waiting once the confirm arrives belongs to that publish.

Result of an attempt	retry	circuit
confirmed	Success	Success
returned unroutable: ErrUnroutable	Stop	Success: the broker answered
a channel error such as 404 NOT_FOUND for a missing exchange, or 403	Stop	Success
nacked by the broker, a dropped connection, a failed dial or handshake	Retry	Failure
ctx's deadline	Expired, by retry	Failure
ctx cancelled by the caller	Expired	Ignored
circuit.ErrRejected	Stop

A retry after a confirm that was lost with its connection can publish the message twice, so publishing is at least once. A consumer of it is idempotent anyway, under queue's contract.

Publish writes a copy of m.Headers with the trace context injected by s.Telemetry.Propagator, so a consumer continues the trace. The caller's map is never changed. Each Publish has a span of kind PRODUCER, named send <exchange>, or send amq.default for the default exchange. It carries the attributes above, with messaging.operation.type and messaging.operation.name both send, and messaging.destination.name as semconv's {exchange}:{routing key}. Each attempt records messaging.client.sent.messages, and messaging.client.operation.duration from the publish to its confirm, each with error.type when it failed.

Running on humus.Main

The Service runs on humus.Main's defaults, a 20s grace period and no drain, which fit a consumer for the reasons the Kafka revision gives. So the package has no App, no Main and no defaults of its own. A consumer next to an API runs on that server package's Main. Two queues, or two Delivery values, are two Services returned from one Init. Like the wave-3 runtimes, the Service serves no health probes. Proposal: exposing health probes, in wave 4, decides how a consumer's pod serves them.

Readiness

Nothing routes traffic to a consumer, so its readiness only gates rollouts. A login, vhost or queue the broker refuses already ends the Service, and the pod's crash loop stops the rollout. A wrong address doesn't: the Service keeps retrying, as it does through a broker's upgrade, and without a monitor its pod turns ready and replaces a pod that consumed. Consuming closes that gap. In the health Service's readiness, it holds a new pod not ready until it has consumed from its queue once. After that, a lost connection doesn't make it unready: a broker outage hits every pod at once, and turning them all unready would change nothing but block the rollout that might fix it.

It needs nothing of humus's but health.MonitorFunc, which landed in wave 1. Proposal: exposing health probes makes the same choice for the Kafka consumer, with Joined.

In use

func Init(ctx context.Context, s *humus.Setup) ([]humus.Service, error) {
	d := rabbitmq.DefaultSettings()
	env := func(name string) config.Value[string] { return config.Env(s.Process.Environment, name) }

	settings, err := config.Struct(func(r *config.Resolver) rabbitmq.Settings {
		k := d
		// No .Default: the address, the login and the queue are required.
		k.Connection.Addr = r.Get(env("RABBITMQ_ADDR").Validate(rabbitmq.CheckAddr))
		k.Connection.SASL = &amqp.PlainAuth{
			Username: r.Get(env("RABBITMQ_USERNAME")),
			Password: r.Get(env("RABBITMQ_PASSWORD")),
		}
		k.Queue = r.Get(env("RABBITMQ_QUEUE").Validate(rabbitmq.CheckQueue))
		k.Prefetch = r.Get(env("RABBITMQ_PREFETCH").Parse(strconv.Atoi).Default(d.Prefetch).Validate(rabbitmq.CheckPrefetch))
		return k
	}).Read(ctx)
	if err != nil {
		return nil, err // exit 78, naming every bad variable
	}

	pub, err := rabbitmq.NewPublisher(s, publisherSettings(settings.Connection))
	if err != nil {
		return nil, err
	}
	h := orders.NewHandler(pub) // idempotent; publishes an order.accepted event

	svc, err := rabbitmq.NewService(s, settings, queue.ProcessorFunc[rabbitmq.Message](h.Handle))
	if err != nil {
		return nil, err
	}
	return []humus.Service{svc}, nil
}

publisherSettings is the end user's helper: DefaultPublisherSettings() with the same Connection.

Design decisions

• No auto-ack, even for AtMostOnce. The Kafka proposal sketched auto-ack, which is strictly at most once. But RabbitMQ ignores the prefetch count for an auto-ack consumer and sends as fast as TCP allows, and amqp091 buffers every delivery without a bound. A backlog then fills the process's memory, and every crash or rollout loses everything buffered, already acked. A batch acked and confirmed before it is processed is just as strict, and bounds the loss to Prefetch, as Kafka's MaxPollRecords does.
• A failure rejects one delivery, and nacks the rest. RabbitMQ counts a reject, a closed channel and a lost connection toward a quorum queue's delivery limit, and not a nack. Leaving the channel to requeue everything would count every prefetched message against the delivery limit whenever a sibling failed, and in a crash loop it would drop innocent messages. Rejecting the failed delivery and nacking the rest counts exactly the message that failed.
• The Service declares nothing. The queue, its exchange, bindings, type, dead-lettering and delivery limit are topology, owned by RabbitMQ's definitions, an operator, or the team that owns the exchange. queue/kafka creates no topics either. A missing queue fails the Service with NOT_FOUND. Open question 2.
• The Service reconnects. See Connection loss, and open question 3.
• Dialled with amqp.Open, not DialConfig. See Connecting. It is what puts the broker on humustest's network in tests, and it keeps credentials out of URLs.
• amqp091's experimental recovery isn't used. It is experimental, it redials through DialConfig's URL, and it redeclares topology. The Service's reconnect is a short loop of retry that humus tests.
• One queue per Service, one consumer per channel. A consumer's acks are on its channel, and its failure stops it, so two queues are two Services, each stopping on its own failure, as two Kafka Delivery values are.
• Concurrency workers on one channel. One channel keeps one prefetch window and one ack stream. Deliveries are handed to workers in delivery order.
• The Service's consumer tag is its own, so amqp091 never reads os.Args[0].
• The Publisher is in this proposal. The charter left it to this proposal. A RabbitMQ processor that publishes, such as story(docs): create walkthrough documentation for a prod ready rabbitmq processor #318's, is the common case, and failure injection in wave 5 needs a RabbitMQ client to add its point to. Open question 4.
• A channel per publish in flight. Correlating a return with its publish any other way means a header of humus's on every message, or trusting the end user's message IDs to be unique.
• Publishing's zero value is persistent. Transient turns persistence off, so a message isn't lost on a broker restart because a field was left out.
• ErrUnroutable is exported. A caller branches on it to fall back, for example to a dead-letter store. It is the only error a caller can act on that isn't amqp091's own *amqp.Error.
• Dependencies. github.com/rabbitmq/amqp091-go v1.15.0, in the nested module, which brings nothing else into the module graph but goleak. bedrock's retry and circuit v0.24.0. testcontainers' RabbitMQ module only behind the build tag.

Testing

A fake broker on humustest's Host

No in-memory RabbitMQ exists for Go. garagemq, a broker in Go, listens with net.ListenTCP and logs through logrus's global logger, stores messages in badger, and has no release since 2024. wabbit's fake is archived, and it fakes a Go interface rather than the wire, so nothing would go through the Host. amqp091's own test server is unexported.

So humus writes one: queue/rabbitmq/rabbitmqtest, a fake AMQP 0-9-1 broker that listens on a humustest Host, as kfake does for Kafka:

package rabbitmqtest // github.com/z5labs/humus/queue/rabbitmq/rabbitmqtest

// NewBroker starts a broker listening on addr through h, with one vhost, "/",
// and a login generated with crypto/rand.Text. t.Cleanup closes it.
func NewBroker(t testing.TB, h *humustest.Host, addr string, opts ...Option) *Broker

func WithTLS(cfg *tls.Config) Option // serves TLS, with a certificate from humustest.NewCA

func (b *Broker) Username() string
func (b *Broker) Password() string

func (b *Broker) DeclareQueue(name string)
func (b *Broker) DeclareExchange(name, kind string) // direct, fanout or topic
func (b *Broker) Bind(queue, exchange, key string)

func (b *Broker) Publish(exchange, key string, p amqp.Publishing) // as a producer would
func (b *Broker) Ready(queue string) []amqp.Delivery              // waiting, in order; for asserting what was published
func (b *Broker) Unacked(queue string) int
func (b *Broker) Returns(queue string) []int // what each waiting message has counted toward a delivery limit
func (b *Broker) CloseConnections(code int, reason string) // as a node going down does: 320 CONNECTION_FORCED

It covers what the Service and the Publisher use: PLAIN logins, channels, basic.qos, consumers, acks, nacks and rejects, requeueing on close, heartbeats both ways, the default exchange, direct, fanout and topic exchanges, publisher confirms and mandatory returns. It counts toward a delivery limit as a quorum queue does: rejects and closes, not nacks. Topology is declared by the test, not over AMQP, since the Service declares none. It has no policies, no TTLs, no dead-lettering and no consumer timeout. Its goroutines are per connection and end with it, so it works inside synctest. No login is fixed: NewBroker generates one, and the test hands it to the App with humustest.WithEnv.

The prototype

A prototype of the Service and a consume-only fake broker, in a scratch module against humus at 4289e89a, amqp091-go v1.15.0 and bedrock retry v0.24.0, passes go test -race -count=3, covering 87.1% of the Service. The fake is about 750 lines: a 200-line codec and a 550-line broker. It has no exchanges, confirms or publishing over AMQP, and the prototype has no telemetry and no Publisher. Its tests run whole Apps with humustest.Start inside synctest, and show:

• In order, then a drain. Five messages are processed in order and acked, through 10 minutes of fake time with heartbeats both ways. With a sixth in flight at SIGTERM, it finishes and is acked, the two prefetched behind it are nacked and ready again, nothing has counted toward a delivery limit, and the queue has no consumer. Exit 0.
• A failure stops the Service and redelivers. Messages 0 to 2 are acked, and 3 fails: exit 1, with the error in the cause. 3 has counted once and 4 not at all, and a second App processes both.
• At most once. Every message is acked, and the ack confirmed, before Process sees it, and a processing error doesn't stop the Service.
• Reconnecting. After the broker closes the connection with 320 CONNECTION_FORCED, the Service reconnects and processes what is published next.
• A missing queue, or a wrong password, ends the App with exit 1, with NOT_FOUND or amqp091's ErrCredentials in the cause.
• TLS. A broker serving a certificate from humustest.NewCA, over a TLS listener on the Host, is consumed as over plaintext.
• The contract. The Service passes humustest.TestService, including the leak check, so amqp091's reader, heartbeater and buffer goroutines end with Serve. The broker is closed by an OnStop function, and, in the other tests, by t.Cleanup, which runs inside the bubble.

A test of the end user's Init, sketched on rabbitmqtest:

func TestInit_at_least_once_stops_on_a_failure_and_redelivers(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		h := humustest.NewHost(t)
		b := rabbitmqtest.NewBroker(t, h, "rabbit:5672")
		b.DeclareQueue("orders")
		for i := range 5 {
			b.Publish("", "orders", amqp.Publishing{MessageId: strconv.Itoa(i)})
		}

		bad := errors.New("cannot process")
		run := func(fail bool) (bedrock.ExitStatus, []string) {
			var seen []string
			p := queue.ProcessorFunc[rabbitmq.Message](func(ctx context.Context, m rabbitmq.Message) error {
				seen = append(seen, m.MessageID)
				if fail && m.MessageID == "3" {
					return bad
				}
				return nil
			})
			proc, status := humustest.Start(t, initWith(p), // the end user's Init, with this Processor
				humustest.WithApp(humus.WithHost(h.Host())),
				humustest.WithEnv(map[string]string{
					"RABBITMQ_ADDR":     "rabbit:5672",
					"RABBITMQ_USERNAME": b.Username(),
					"RABBITMQ_PASSWORD": b.Password(),
					"RABBITMQ_QUEUE":    "orders",
				}),
			)
			if !fail {
				synctest.Wait() // every message is processed and acked
				proc.Signal(syscall.SIGTERM)
			}
			return <-status, seen
		}

		st, seen := run(true)
		require.Equal(t, humus.ExitFailure, st.Code())
		require.ErrorIs(t, st.Cause(), bad)
		require.Equal(t, []string{"0", "1", "2", "3"}, seen)
		require.Equal(t, []int{1, 0}, b.Returns("orders"), "3 was rejected and 4 was nacked")

		st, seen = run(false)
		require.Zero(t, st.Code(), st.Cause())
		require.ElementsMatch(t, []string{"3", "4"}, seen)
	})
}

The stories add what the prototype leaves out:

• every check, NewService and NewPublisher naming every bad field, and DefaultSettings and DefaultPublisherSettings passing them with the required fields set;
• the span's parent is a producer's trace context from the headers, its attributes and the two instruments, and the Publisher's span, instruments and injected headers, all on humustest.Telemetry;
• Concurrency above 1 processes deliveries at once, and 1 keeps their order;
• Consuming is false before the broker's first basic.consume-ok, and while the address is unreachable, true after it, through a reconnect, and false once Serve has returned;
• a grace period that runs out during a slow Process exits 1 with a *humus.GracePeriodError;
• the Publisher: a confirmed publish, ErrUnroutable without a retry, a retry after the broker closes the connection, the breaker opening after 5 failures in a row and logging at WARN, and a Publish cut short by its context;
• rabbitmqtest's own tests, and a runnable Example.

Container tests

Tests against RabbitMQ in a container, behind the testcontainers build tag, run in a CI job of their own, as queue/kafka's do. They don't count toward the coverage gate. They check what the fake can only claim:

• a quorum queue's delivery count after a failure, a shutdown and a crash;
• that a basic.qos reply follows the ack before it;
• requeue order;
• returns and confirms;
• 320 CONNECTION_FORCED from a broker that stops.

Migration from today

humus has no RabbitMQ package to migrate from. For a consumer written on amqp091-go directly:

On amqp091-go	With this proposal
amqp.Dial(os.Getenv("AMQP_URL"))	Settings.Connection, sourced in Init, dialled through s.Host
ch.Qos(n, 0, false)	Settings.Prefetch
ch.Consume(q, "", autoAck, ...) and a loop over the deliveries	Settings.Queue, Settings.Delivery, and a queue.Processor[rabbitmq.Message]
d.Ack(false), d.Nack(false, true)	the Service, as Delivery says
conn.NotifyClose and a reconnect loop	the Service
ch.PublishWithContext	Publisher.Publish, with confirms, returns, retries and a breaker
amqp.SetLogger, OTel's globals	s.Telemetry

Two things change for such a consumer: a processing error under at-least-once stops the App, and auto-ack is no longer how at-most-once is done.

Open questions

1. How are RabbitMQ consumers tested?

   • (a) rabbitmqtest, a fake broker humus writes and keeps. Every App that uses RabbitMQ runs whole in synctest, as the charter's principle 6 requires. Fidelity is humus's to keep, and the container tests catch where it drifts.
   • (b) Containers only. They are the real broker, but no RabbitMQ App runs in synctest, every test starts a broker, CI needs Docker, and the coverage gate can't count them.
   • (c) No broker: unit tests against an interface over amqp091's channel. They are cheap, but an end user can't test their whole App.

   Recommendation: (a), with container tests in their own CI job, as for Kafka.

2. Topology.

   • (a) The Service declares nothing, and a missing queue stops it with NOT_FOUND, as queue/kafka creates no topics.
   • (b) Settings declare the queue and its bindings. That grows into RabbitMQ's whole topology API: queue types, arguments, exchanges, dead-lettering.
   • (c) A Declare func(context.Context, *amqp.Channel) error field, called on each connection before consuming. It gives full control, at the price of putting amqp091's Channel in humus's API.

   Recommendation: (a).

3. A lost connection.

   • (a) The Service reconnects with backoff, as queue/kafka keeps retrying. A RabbitMQ rolling upgrade closes every connection with 320 CONNECTION_FORCED.
   • (b) Serve returns the error and the pod restarts. That is louder and has no reconnect code, but every consumer pod restarts on every broker upgrade, with Kubernetes' crash-loop back-off of up to 5 minutes once restarts repeat.

   Recommendation: (a).

4. The Publisher.

   • (a) Keep it in this proposal, as proposed.
   • (b) Consume only. Publishing goes out of scope: the end user publishes with amqp091 over s.Host.Dial, without humus's telemetry, retries or breaker, and wave 5's failure injection has no RabbitMQ client to add its point to.

   Recommendation: (a).

Out of scope

• Declaring queues, exchanges and bindings, and RabbitMQ policies: topology, unless open question 2 decides otherwise.
• Dead letters: the queue's dead-letter exchange and delivery limit, or the Processor's own store. The contract has no way to dead-letter a message on purpose.
• RabbitMQ streams, super streams, AMQP 1.0, basic.get, consumer priorities and arguments, and direct reply-to.
• More than one broker address, and failing over between them: Addr is one address, such as a DNS name or a load balancer in front of the cluster.
• OAuth 2 tokens refreshed on an open connection (connection.update-secret). A token is a PlainAuth password, read at each connection.
• Health probes: Proposal: exposing health probes (wave 4) decides how a consumer's pod serves them.
• Stripping failure-injection scenarios from deliveries, between header extraction and Process, and the Publisher's injection point: Proposal: failure injection (wave 5).
• The RabbitMQ processor walkthrough (story(docs): create walkthrough documentation for a prod ready rabbitmq processor #318): Proposal: walkthroughs and a humus MIGRATING.md (wave 6).

What the stories change outside the module:

• .github/workflows/build.yaml gets a "RabbitMQ (testcontainers)" job, which isn't a required check, and a step that vets the tagged tests. .claude/backlog.json needs nothing: .github/scripts/go-modules.sh finds the new module.
• .golangci.yaml gets forbidigo's rule against amqp091's SetLogger.
• The docs site gets a new page, "Consuming RabbitMQ", next to "Consuming Kafka". The queue package doc names queue/rabbitmq beside queue/kafka, and the testing page covers rabbitmqtest. The root package doc's "Being rebuilt" says RabbitMQ has landed. The lifecycle page's ambient leaks gain amqp091's package-level logger.

References

• Proposal: humus on the bedrock process model, revision 7: RabbitMQ in wave 4, and the conventions of the wave-4 clients
• Proposal: queue processing and the Kafka runtime and its revision; queue and queue/kafka as landed; queue/kafka: a fenced-out member's late commit can move the new owner's committed offset #757
• Proposal: humus.Main
• story(queue): implement a runtime for rabbitmq #310 (RabbitMQ), story(docs): create walkthrough documentation for a prod ready rabbitmq processor #318 (a RabbitMQ processor walkthrough)
• amqp091-go v1.15.0: Open and DialConfig, the heartbeater, Qos, Consume, the consumer buffer
• RabbitMQ: quorum queues' poison message handling, consumer acknowledgements and publisher confirms, consumer prefetch, consumers
• OpenTelemetry's semantic conventions for RabbitMQ, v1.43.0
• bedrock retry and circuit v0.24.0

[Zaba505]

Revision 1: the open questions answered

The open questions:

1. Tests that need a broker run against RabbitMQ itself, in a container, and humus writes no fake broker. humus is neither RabbitMQ's client nor its server, so it doesn't write or keep a broker implementation. rabbitmqtest is dropped, along with the prototype's fake. Everything that doesn't need a broker is still unit tested: the settings, the pure functions, and the Service's consume loop and the Publisher's attempts, on an unexported stand-in for amqp091's channel that implements no part of AMQP. So no App that uses RabbitMQ runs whole inside synctest, which departs from the charter's principle 6. The charter's revision 10 records the exception.
2. The Service declares nothing, as recommended. The end user's infrastructure is fully configured and initialized before the Service starts. The Service is purely a consumer, and the Publisher only publishes to exchanges that already exist.
3. The Service reconnects with backoff, as recommended.
4. The Publisher stays in this proposal, as recommended.

Three decisions made in other wave-4 proposals reach this one, and this revision records them:

• Exposing health probes' revision 2 gives the Kafka consumer Joined() health.Monitor, for readiness only, and keeps it in step with Consuming. So Consuming becomes func (svc *Service) Consuming() health.Monitor, and it is for readiness only too. That revision also settles how a consumer's pod serves its probes.
• Outbound HTTP clients' revision 1 lists what reaches the other clients. A nil Connection.TLS stays plain text, for the reason it gives: humus can't control what a service connects to. The Publisher gets no Classify field: a broker's protocol, not an API, fixes what its answers mean, so its classification stays fixed. And it records no instruments of its own.
• The charter's revision 9 makes instruments of humus's own a proposal in wave 5, Proposal: humus's own metrics. This proposal's references now name it.

Each section below replaces or adds to the part of the same name in the post. Background, Connecting, Settings, What Serve does, Delivery, Shutdown, Connection loss, Failures, Telemetry, In use, Migration from today and References are unchanged.

Summary

The bullets on Consuming and on tests become, and a bullet on topology is added after the one on dialling:

• The Service declares nothing. The queue, its exchange, bindings, type, dead-lettering and delivery limit are the end user's infrastructure, set up before the Service starts. A missing queue stops the Service with NOT_FOUND.
• Consuming() returns a readiness monitor. It is unhealthy until the Service first consumes its queue, so a pod with a wrong broker address never turns ready, and healthy after that. It is for readiness only.
• Tests that need a broker run against RabbitMQ in a container. humus writes no fake broker. Unit tests, inside synctest, cover the settings, the pure functions, and the consume loop and the Publisher's attempts on an unexported stand-in for amqp091's channel. Container tests run whole Apps against RabbitMQ, in a CI job of their own, as for Kafka. An end user tests their Processor as a unit, and their whole App against a container.

Proposed API

Consuming becomes:

// Consuming returns a monitor that is healthy once the Service has started
// consuming its queue: unhealthy until the broker first answers its
// basic.consume, and healthy from then until Serve returns. A lost
// connection after that doesn't make it unhealthy. It never returns an
// error. A consumer is made ready once it consumes with
//
//	health.Readiness(consumer.Consuming())
//
// It is for readiness only: restarting the pod doesn't bring the broker
// back, and an error that a restart would fix already ends Serve.
func (svc *Service) Consuming() health.Monitor

queue/rabbitmq imports health, which imports nothing but the standard library.

Order and concurrency

The last sentence becomes:

A Processor that needs per-key order uses a single-active-consumer queue, which is topology the end user sets up, and Concurrency 1.

The Publisher

In the second bullet, "humus adds no metrics of its own for them" becomes:

humus records no instruments of its own for them. Whether it should, for every client at once, is Proposal: humus's own metrics (wave 5).

This is added after the table of attempt results:

The table is fixed. The HTTP client's Classify exists because each API decides what its answers mean. RabbitMQ's protocol fixes what a confirm, a nack, a return and a channel error mean, so the Publisher has no Classify field, as producing to Kafka's Producer has none.

Running on humus.Main

The last two sentences become:

The Service serves no health probes of its own. A consumer's pod serves them on a probes server, as Readiness shows.

Readiness

This section is replaced:

Nothing routes traffic to a consumer, so its readiness only gates rollouts. A login, vhost or queue the broker refuses already ends the Service, and the pod's crash loop stops the rollout. A wrong address doesn't: the Service keeps retrying, as it does through a broker's upgrade, and without a monitor its pod turns ready and replaces a pod that consumed. Consuming closes that gap. In a probes server's readiness, its monitor holds a new pod not ready until it has consumed from its queue once. After that, a lost connection doesn't make it unhealthy: a broker outage hits every pod at once, and turning them all unready would change nothing but block the rollout that might fix it.

Consuming is for readiness only. A liveness monitor would restart every pod in a broker outage. The kubelet's restart backoff, up to five minutes by default, would then hold them down after the broker comes back. The Service reconnects on its own, and the errors a restart would fix already end Serve, so the pod exits and is restarted without a probe.

A consumer's pod serves its probes on httphealth.NewService, a probes server of its own, as exposing health probes has a Kafka consumer's pod do:

func Init(ctx context.Context, s *humus.Setup) ([]humus.Service, error) {
	settings, probeSettings, err := readSettings(ctx, s) // rabbitmq.Settings and httphealth.Settings
	if err != nil {
		return nil, err
	}
	consumer, err := rabbitmq.NewService(s, settings, queue.ProcessorFunc[rabbitmq.Message](h.Handle))
	if err != nil {
		return nil, err
	}
	probes, err := httphealth.NewService(s, probeSettings, health.Readiness(consumer.Consuming()))
	if err != nil {
		return nil, err
	}
	return []humus.Service{consumer, probes}, nil
}

Design decisions

The bullets "The Service declares nothing", "The Service reconnects", "Dialled with amqp.Open, not DialConfig", "The Publisher is in this proposal" and "Dependencies" become:

• The Service declares nothing. The end user's infrastructure is fully configured and initialized before the Service starts. The queue, its exchange, bindings, type, dead-lettering and delivery limit are owned by RabbitMQ's definitions, an operator, or the team that owns the exchange. The Service is purely a consumer, and the Publisher publishes only to exchanges that exist. queue/kafka creates no topics either. A missing queue stops the Service with NOT_FOUND, and a missing exchange fails a publish at its first attempt with 404 NOT_FOUND.
• The Service reconnects. See Connection loss. A rolling upgrade of RabbitMQ closes every connection with 320 CONNECTION_FORCED. If Serve returned instead, every consumer pod would restart on every broker upgrade, and repeated restarts put a pod in Kubernetes' crash-loop backoff, of up to five minutes.
• Dialled with amqp.Open, not DialConfig. See Connecting. It dials through s.Host, so the connecting code is unit tested on humustest's Host and the container tests reach RabbitMQ through humus.OSHost(). It also keeps credentials out of URLs.
• The Publisher is in this proposal. The charter left it to this proposal. A RabbitMQ processor that publishes, such as story(docs): create walkthrough documentation for a prod ready rabbitmq processor #318's, is the common case, and failure injection in wave 5 needs a RabbitMQ client to add its point to.
• Dependencies. github.com/rabbitmq/amqp091-go v1.15.0, in the nested module, which brings nothing else into the module graph but goleak. bedrock's retry and circuit v0.24.0. Behind the build tag only: testcontainers' RabbitMQ module, which every test that needs a broker runs on, and goleak.

These are added:

• No fake broker. humus is neither RabbitMQ's client nor its server, so it doesn't write or keep a broker. A fake's fidelity would be humus's to maintain against every RabbitMQ release, and its tests would show what the fake does rather than what RabbitMQ does. Tests that need a broker run against RabbitMQ itself, and everything else is a unit test. The cost is that no App that uses RabbitMQ runs whole inside synctest: the charter's revision 10 records that exception to principle 6.
• A stand-in channel for the unit tests. The consume loop and the Publisher's attempts use their channel through an unexported interface, which *amqp.Channel satisfies, so their logic is unit tested inside synctest. The stand-in implements no part of AMQP and keeps no queue. It hands out the deliveries a test gives it, and records what the Service does with them. It is unexported, so none of it is API.
• Consuming returns a health.Monitor. health.Readiness(consumer.Consuming()) needs no conversion, and the type says what the method is for. It has the Kafka consumer's Joined shape. It is API because the end user has no other way to learn that the Service consumes.
• Plain text is allowed. A nil Connection.TLS is plain text, as before. humus makes a service as secure as it can, but it can't control what the service connects to.
• No Classify. See The Publisher.

Testing

This section is replaced.

No in-memory RabbitMQ exists for Go, and humus doesn't write one. So each test runs either with no broker, as a unit test, or against RabbitMQ itself, in a container. The prototype's results stand as evidence for the design, and the container tests now show each of them against RabbitMQ.

Unit tests

They run with the module's other tests in Build, inside testing/synctest wherever time matters, and they count toward the coverage gate. None of them speaks AMQP to a broker.

• The settings. Every check, and NewService and NewPublisher naming every bad field. They also reject a nil Processor, a Concurrency above Prefetch and a BreakerPolicy without Trip or Probe. DefaultSettings and DefaultPublisherSettings pass every check once the required fields are set.
• The pure functions:
  • a Message built from an amqp.Delivery;
  • an amqp.Publishing built from a Publishing, whose zero value is persistent;
  • the header carrier, which reads string values only;
  • the copy of Headers that Publish injects the trace context into, with the caller's map unchanged;
  • semconv's destination name, the span names and their attributes;
  • which errors stop the Service and which it reconnects after;
  • the Publisher's outcomes for retry and circuit.
• The consume loop and the Publisher's attempts, on the stand-in. The Service and the Publisher get their channels through an unexported function, and a test replaces it with one that returns the stand-in. The stand-in hands the Service the deliveries the test gives it, and records each Ack, Nack, Reject and Qos. It confirms, nacks or returns each publish as the test says, and closes as a lost connection would. The tests cover:
  • under AtLeastOnce, the ack once Process returns nil. On an error: one reject, the other workers' deliveries acked, one Nack with multiple for the rest, and Serve's error naming the queue and the message's ID;
  • under AtMostOnce, one Ack with multiple, and the Qos after it, before any Process. An error is logged at ERROR and counted, and processing goes on;
  • Concurrency 1 keeping the delivery order, and 4 processing four deliveries at once;
  • shutdown's steps, and a grace period that runs out during a slow Process, which cancels its context and exits 1 with a *humus.GracePeriodError;
  • a lost connection. The acks that fail after it are dropped, it is logged at WARN with its code and reason, and the Service reconnects. On an error a reconnect doesn't fix, the Service stops instead;
  • Consuming, which is unhealthy before the first consume succeeds, healthy from then and through a reconnect, and unhealthy once Serve has returned;
  • the span per delivery, its parent from the headers, its attributes, and the two instruments, on humustest.Telemetry;
  • the Publisher:
    • a confirmed publish;
    • ErrUnroutable with no retry;
    • a nack retried;
    • a channel lost mid-publish, retried on a new one;
    • the breaker opening after 5 failures in a row, with its WARN record;
    • a Publish cut short by its context;
    • its span, instruments and injected headers.
• The connecting code, on humustest's Host, with no broker:
  • With nothing listening at Addr, Serve retries with backoff and logs each attempt at WARN. Consuming stays unhealthy, and Serve returns nil once stopped.
  • With a listener that accepts and never answers, each attempt ends at ConnectTimeout.
  • Against a TLS listener with a certificate from humustest.NewCA, the handshake sends Addr's host as the server name when TLS.ServerName is empty.
  • With nothing listening for the Publisher, each failed dial is retried and counts as a breaker failure.
• The contract. The Service passes humustest.TestService with nothing listening at Addr, so it shuts down as the lifecycle expects while it waits for a broker.

What runs only against a broker is the code that opens AMQP over the connection, opens a channel and hands it to the code above. The container tests cover it.

Container tests

The container tests are behind the testcontainers build tag. They run the official rabbitmq image, at a 4.x release the tests pin, with testcontainers' RabbitMQ module. Their Apps run on humus.OSHost(), in real time, as queue/kafka's container tests do. Each test declares its own queues over its own amqp091 connection, since the Service declares none. The broker's login is generated with crypto/rand.Text, never the image's guest. They show against RabbitMQ what the prototype showed against its fake, and what only a broker can:

• In order, then a drain. Five messages are processed in order and acked. A sixth is in flight at SIGTERM, and it finishes and is acked. The messages prefetched behind it are ready again, in their order, and none has counted toward the delivery limit. The queue has no consumer, and the exit code is 0.
• A failure stops the Service and redelivers. On a quorum queue, the message that failed comes back with x-delivery-count 1, and the one nacked behind it with none. A second App processes both.
• A grace period that runs out. The connection closes with deliveries unacked, and each one counts once toward the delivery limit.
• At most once. The broker closes the connection, with rabbitmqctl close_all_connections, while Process blocks on a batch's first message. No message of the batch is delivered again, so the basic.qos reply came after the broker had the ack.
• Reconnecting. rabbitmqctl close_all_connections, run in the container, closes the Service's connection with 320 CONNECTION_FORCED. The Service logs it at WARN, Consuming stays healthy, the Service reconnects, and it processes what is published next.
• Refusals. A missing queue ends the App with exit 1 and NOT_FOUND in the cause. So does a wrong password, with amqp091's ErrCredentials.
• TLS. The test writes a certificate from humustest.NewCA to its temp directory and mounts it into the container. RabbitMQ serves it, and the Service consumes and the Publisher publishes as over plain text.
• The Publisher:
  • A confirmed publish reaches its queue with the trace context in its headers, and the consumer's span continues the trace.
  • An unbound routing key returns ErrUnroutable, without a retry.
  • A missing exchange fails at the first attempt, with 404 NOT_FOUND.
  • A publish across close_all_connections is retried on a new connection and confirmed.
• Nothing left behind. Once Serve returns, none of amqp091's goroutines is still running, as goleak checks.

In CI

• A job named "RabbitMQ (testcontainers)" runs go -C queue/rabbitmq test -race -tags testcontainers ./..., as Kafka's job does. It isn't a required check. The backlog loop waits for every check, so no story merges past a failure in it. Build vets the tagged tests without starting a container.
• The coverage gate counts the unit tests alone, as it does for every module. The unit tests above are what meet it.

Testing the end user's App

• Its Processor, as a unit. A queue.Processor[rabbitmq.Message] is a function of a plain struct. So service code is tested by calling it with Message values, inside synctest, with its other dependencies, such as HTTP, gRPC or Kafka, on humustest's Host. Service code that publishes takes an interface of its own over Publish, which its unit tests fake.
• The whole App, against a container. Its test runs the App with humustest.Start and humus.WithHost(humus.OSHost()), against RabbitMQ in a container. The App still reads its environment from humustest's process. This is the test that the charter's principle 6, as revision 10 amends it, asks this proposal to show. The docs page shows it:

//go:build testcontainers

func TestInit_at_least_once_stops_on_a_failure_and_redelivers(t *testing.T) {
	user, password := rand.Text(), rand.Text() // crypto/rand: no fixed login
	c, err := tcrabbitmq.Run(t.Context(), rabbitImage,
		tcrabbitmq.WithAdminUsername(user),
		tcrabbitmq.WithAdminPassword(password),
	)
	testcontainers.CleanupContainer(t, c)
	require.NoError(t, err)
	addr, err := c.PortEndpoint(t.Context(), "5672/tcp", "")
	require.NoError(t, err)

	// The queue is the end user's infrastructure, so the test declares it.
	ch := openChannel(t, c) // amqp.Dial on c.AmqpURL; t.Cleanup closes it
	_, err = ch.QueueDeclare("orders", true, false, false, false,
		amqp.Table{amqp.QueueTypeArg: amqp.QueueTypeQuorum})
	require.NoError(t, err)
	for i := range 5 {
		err := ch.PublishWithContext(t.Context(), "", "orders", false, false,
			amqp.Publishing{MessageId: strconv.Itoa(i)})
		require.NoError(t, err)
	}

	bad := errors.New("cannot process")
	run := func(fail bool) (bedrock.ExitStatus, []string) {
		var mu sync.Mutex
		var seen []string
		p := queue.ProcessorFunc[rabbitmq.Message](func(ctx context.Context, m rabbitmq.Message) error {
			mu.Lock()
			defer mu.Unlock()
			seen = append(seen, m.MessageID)
			if fail && m.MessageID == "3" {
				return bad
			}
			return nil
		})
		proc, status := humustest.Start(t, initWith(p), // the end user's Init, with this Processor
			humustest.WithApp(humus.WithHost(humus.OSHost())),
			humustest.WithEnv(map[string]string{
				"RABBITMQ_ADDR":     addr,
				"RABBITMQ_USERNAME": user,
				"RABBITMQ_PASSWORD": password,
				"RABBITMQ_QUEUE":    "orders",
			}),
		)
		if !fail {
			require.Eventually(t, func() bool {
				mu.Lock()
				defer mu.Unlock()
				return len(seen) == 2
			}, 30*time.Second, 10*time.Millisecond)
			proc.Signal(syscall.SIGTERM) // the drain acks what is in flight
		}
		st := <-status
		mu.Lock()
		defer mu.Unlock()
		return st, slices.Clone(seen)
	}

	st, seen := run(true)
	require.Equal(t, humus.ExitFailure, st.Code())
	require.ErrorIs(t, st.Cause(), bad)
	require.Equal(t, []string{"0", "1", "2", "3"}, seen)

	st, seen = run(false)
	require.Zero(t, st.Code(), st.Cause())
	require.ElementsMatch(t, []string{"3", "4"}, seen)
}

tcrabbitmq is testcontainers' RabbitMQ module, and rabbitImage a pinned 4.x rabbitmq image. openChannel and initWith are the end user's helpers.

What waves 5 and 6 build on this

This section is new. Both proposals named in the note at the top get RabbitMQ without a fake broker:

• Proposal: failure injection (wave 5) strips scenarios at the Service and adds an injection point to the Publisher's attempt. Its RabbitMQ tests are unit tests on the stand-in, and container tests where they need a broker.
• Proposal: walkthroughs and a humus MIGRATING.md (wave 6) builds story(docs): create walkthrough documentation for a prod ready rabbitmq processor #318's RabbitMQ processor on this. Its app_test.go runs the App against RabbitMQ in a container, on humus.OSHost(), rather than inside synctest. Test cases written once still run both there and against a deployed environment.

Open questions

All four are answered above. None are left.

Out of scope

The bullets on topology and on health probes become:

• Declaring queues, exchanges and bindings, and RabbitMQ policies: topology, which the end user's infrastructure sets up before the Service starts.
• A probes server of the Service's own: a consumer's pod serves its probes with httphealth.NewService, from exposing health probes (wave 4), as Readiness shows.

These are added:

• A fake RabbitMQ broker, for humus's tests or for end users'.
• Instruments of humus's own, such as gauges of the Publisher's breaker state and its retry budget's tokens: Proposal: humus's own metrics (wave 5).

The list of what the stories change outside the module becomes:

• .github/workflows/build.yaml gets a "RabbitMQ (testcontainers)" job, which isn't a required check, and Build's step that vets the tagged tests vets queue/rabbitmq's too. .claude/backlog.json needs nothing: .github/scripts/go-modules.sh finds the new module.
• .golangci.yaml gets forbidigo's rule against amqp091's SetLogger.
• The docs site gets a new page, "Consuming RabbitMQ", next to "Consuming Kafka". It covers a consumer's probes and both kinds of test above. The queue package doc names queue/rabbitmq beside queue/kafka. The testing page says how an App that uses RabbitMQ is tested, under principle 6's exception. The root package doc's "Being rebuilt" says RabbitMQ has landed. The lifecycle page's ambient leaks gain amqp091's package-level logger.

[Zaba505]

Closing: RabbitMQ has landed

The revision's stories are done:

• story(rabbitmq): configure a RabbitMQ consumer with Settings and a check per field #807 configured a RabbitMQ consumer with Settings and a check per field (story(rabbitmq): configure a RabbitMQ consumer with Settings and a check per field #819);
• story(rabbitmq): connect to RabbitMQ through the Host, and test against RabbitMQ in a container, in a CI job of its own #808 connected to RabbitMQ through the Host, with forbidigo's rule against amqp091's SetLogger, and tested against RabbitMQ in a container in a "RabbitMQ (testcontainers)" job of its own (story(rabbitmq): connect to RabbitMQ through the Host, and test against RabbitMQ in a container, in a CI job of its own #824);
• story(rabbitmq): consume a RabbitMQ queue as a humus Service, at least or at most once #809 consumed a queue as a humus Service, at least or at most once (story(rabbitmq): consume a RabbitMQ queue as a humus Service, at least or at most once #832);
• story(rabbitmq): reconnect after a lost connection, and stop on what a reconnect can't fix #810 reconnected after a lost connection or a consumer the broker cancelled, and stopped on what a reconnect can't fix: a refusal, or a 406 after consuming started (story(rabbitmq): reconnect after a lost connection, and stop on what a reconnect can't fix #844);
• story(rabbitmq): hold a consumer's pod not ready until it consumes, with Consuming #811 held a consumer's pod not ready until it consumes, with Consuming() health.Monitor (story(rabbitmq): hold a consumer's pod not ready until it consumes, with Consuming #846);
• story(rabbitmq): trace and measure every delivery with the Setup's telemetry #812 traced and measured every delivery with the Setup's telemetry (story(rabbitmq): trace and measure every delivery with the Setup's telemetry #848);
• story(rabbitmq): publish with confirms through rabbitmq.Publisher #813 published with confirms through rabbitmq.Publisher (story(rabbitmq): publish with confirms through rabbitmq.Publisher #850);
• story(rabbitmq): trace and measure every publish, and carry the trace to the consumer #814 traced and measured every publish, and carried the trace to the consumer in the message's headers (story(rabbitmq): trace and measure every publish, and carry the trace to the consumer #853);
• story(rabbitmq): retry a failed publish within a budget #815 retried a lost publish within a retry budget (story(rabbitmq): retry a failed publish within a budget #855);
• story(rabbitmq): a circuit breaker fails publishes at once while the broker keeps failing #816 put a circuit breaker in front of publishes while the broker keeps failing, with Breaker and BreakerPolicy (story(rabbitmq): a circuit breaker fails publishes at once while the broker keeps failing #857);
• story(docs): document consuming and publishing with RabbitMQ #817 documented consuming and publishing with RabbitMQ, on a new page that follows Kafka's (story(docs): document consuming and publishing with RabbitMQ #859).

The revision left no open questions. Tests that need a broker run against RabbitMQ in a container, and humus writes no fake broker, as the charter's revision 10 records. The Service declares nothing, it reconnects with backoff, and the Publisher stayed in this proposal.

The rest is out of scope or belongs to other proposals, as "Out of scope" lists:

• Stripping failure-injection scenarios from deliveries, and the Publisher's injection point, are Proposal: failure injection (wave 5).
• Instruments of humus's own for the Publisher's breaker and retry budget are Proposal: humus's own metrics (wave 5), as the charter's revision 9 places them.
• The RabbitMQ processor walkthrough (story(docs): create walkthrough documentation for a prod ready rabbitmq processor #318) is Proposal: walkthroughs and a humus MIGRATING.md (wave 6).
• Declaring topology and policies; dead letters; streams, super streams, AMQP 1.0, basic.get, consumer priorities and arguments, and direct reply-to; more than one broker address and failing over; OAuth 2 tokens refreshed on an open connection; a probes server of the Service's own; and a fake RabbitMQ broker stay out of scope.