Internals: RabbitMQ

How @opentf/esrun-rabbitmq is put together, and why each rule it enforces is there. The decision and what was rejected are in docs/DECISIONS.md (D171).

A package, not a runtime module

Everything the client needs is already in the runtime: runtime:net opens a TCP or TLS socket and hands back web streams. So the client is JavaScript over that API, as the database drivers are — no new code in esrun, and no new capability: reaching the broker is --allow-net, scoped to its host like any other socket.

That also decides what it cannot do. It reaches only hosts the run was granted, and TLS is verified by the runtime's own verifier, with no option here to switch it off. It speaks AMQP 0-9-1, the protocol RabbitMQ implements natively; AMQP 1.0, the Streams protocol, and the management HTTP API are different addressing models and are out of scope.

The layers

Each layer is usable alone; the ones below it decide what the ones above can assume. A delivery is a method frame, a content header, and one or more body frames, so it is the channel's job to reassemble it — the connection only routes frames by channel number.

Framing, and why a lost socket is fatal

A frame is a type, a channel number, a length, the payload, and an end byte. On a stream, a read can end anywhere, so a resumable decoder accumulates bytes and yields a frame only when it is complete.

If the socket is lost, or a frame is not AMQP, the stream is at an unknown boundary and every later frame would be misread. So such a failure is latched once, and every caller — now and later — gets that one cause rather than a new symptom of the same dead connection. There is no resynchronisation: a stream that lost a byte cannot be repaired, only replaced.

A request's answer is awaited before it is asked for

A method and its answer are separate frames, and the answer can be read and routed before the write that carried the request has even settled — a socket under backpressure resolves a write late. An answer that finds nobody waiting is dropped. So every request registers its waiter first and sends second, in the handshake, in channel open and close, and on every channel method.

Answers are paired with requests by order, per channel. That makes a timeout more than a slow call: once one answer is missing, a late arrival would settle the next request of its kind — a second queueDeclare resolving with the first one's queue. So a request that times out closes its channel, with the timeout as the reason, rather than leaving it usable and quietly mispaired. A channel.open that times out keeps its number until the broker answers the channel.close that follows, because a broker that opened it late would refuse a second channel.open on the same number with 504 CHANNEL_ERROR, a connection error.

Why writes are serialised

AMQP lets a connection carry many channels over one socket. A channel's frames may not overtake one another, and a content message's method, header and body frames must arrive together — a method for another publish that slipped between them is what RabbitMQ refuses with 505 UNEXPECTED_FRAME.

Writing each frame as it is asked for is not enough, because two publishes issued at once would interleave halfway. So a connection's writes — the protocol header aside — go through one queue, and a whole content message is one unit on it. A publish takes its turn, writes its three frames, and the next waits. The cost is that a connection's throughput is serialised; the alternative is a broker that rejects the message.

Heartbeats

The heartbeat interval is negotiated under the specification's rule: a 0 on either side means "no proposal", so the greater of the two is used, and 0 on both means none. A quiet connection sends a heartbeat frame on the interval and expects traffic before twice it, or the broker will close it.

The heartbeat timer is unreferenced: it does not keep a script alive that has otherwise finished, so a program exits when its work is done rather than waiting out the interval.

Confirms and acknowledgements

They are opposite directions and use separate counters. A publisher confirm is the broker telling a publishing channel that it has taken the message; RabbitMQ numbers those from 1 per channel. An acknowledgement is a consuming channel telling the broker that a delivery is handled; the broker numbers those from 1 per channel. One message can be both, and the two numbers have nothing to do with each other.

Confirm numbering starts at the confirm.select itself, not at its answer: a publish issued between the two is already the broker's number 1. So confirm mode begins when the request is sent; begun at the answer, that publish went untracked and every later confirm settled the wrong message.

A blocked broker is waited out, not written into

During a memory or disk alarm RabbitMQ sends connection.blocked and stops reading the publishing socket, so "publish operations will eventually time out or fail outright" (RabbitMQ's alarms documentation). Writing into it anyway only grows the connection's write queue in this process. So a publish that starts while the connection is blocked waits at the client until connection.unblocked — the hook is still there, but the waiting no longer depends on the application wiring it up. The sequence number a confirm will use is taken only when the frames are queued, so a publish that waited cannot shift the numbering.

Every publish has one deadline over the whole of it — the wait, the write, and in confirm mode the confirm — defaulting to the channel's timeout. Nothing in AMQP bounds a confirm, and a publish awaiting one that never came waited forever. A timed-out publish's fate at the broker is unknown, as RabbitMQ's confirm documentation says of any unconfirmed message; it is reported as a timeout, not retried, because a retry could duplicate it.

Awaited confirms change what publish means: in confirm mode it resolves once the broker has taken the message and rejects when it cannot, so awaiting the publish awaits the message's fate — the alternative is a publish that resolved before anything had happened to it.

A subscription is never unbounded by default

RabbitMQ's prefetch defaults to no limit, and a subscription buffers every delivery its iterator has not read yet, so a consumer that subscribed to a deep queue without calling qos() first received the whole queue into memory. Most clients inherit that default; this one does not. A subscription sets a prefetch of 250 when neither it nor its channel names one — inside RabbitMQ's recommended 100–300, and Spring AMQP's own default. A prefetch the application set with channel.qos() is left alone, and prefetch: 0 still asks for unlimited. Each subscription's basic.qos and basic.consume go out as a pair, in turn, because RabbitMQ applies the prefetch in force when the consume arrives.

Cancelling hands back what was never read

basic.cancel "does not affect already delivered messages" (AMQP 0-9-1): a delivery that reached the client but was not read when the consumer was cancelled stays unacknowledged, held for that channel until it closes — on a long-lived channel, indefinitely. So a cancel rejects each unread delivery back to its queue, and one that raced the cancel too; the consumer's tag stays routed until cancel-ok so a racing delivery is seen. A noAck subscription has nothing to hand back.

Request/reply, without a reply queue

The usual RPC shape — the caller declares a private queue, publishes a request naming it, and drains it — costs a queue per caller and a routing step for every reply. RabbitMQ's direct reply-to removes both: a channel that consumes amq.rabbitmq.reply-to gets a private pseudo-queue the broker rewrites onto every request's reply_to, so the reply comes straight back to that channel. The consumer has to be noAck, and it has to be started before the first call goes out, or a fast reply arrives with nothing listening — which is why the RPC layer owns the consumer and opens it before it returns. The consumer is the reply path, so a broker that cancels it — a variant that does not, a policy that does — has ended the conversation: the client closes and the calls in flight fail, rather than waiting out a timeout for an answer that can no longer come.

Replies are matched by the correlation id the client generates, not by order, so concurrent calls on one channel are told apart and a stray or duplicated reply is dropped rather than mistaken for the wrong answer.

A URL parameter is honoured or refused

RabbitMQ's own clients read heartbeat, connection_timeout, channel_max, auth_mechanism and a set of TLS parameters from the URL query, and this client reads the ones it can honour (plus frame_max). The rest are refused rather than ignored. cacertfile, certfile and keyfile name files on disk, and a URL is not allowed to make the process read one — the ca setting takes the PEM itself, read under the caller's own grant. verify=verify_none asks for no certificate check, which runtime:net does not offer. And a misspelt parameter that is silently dropped is a setting the caller believes is in force.

Opening the socket is bounded too (connectTimeout, by default the method timeout): a host that drops packets would otherwise hold openConnection for as long as the operating system retries.

Recovery is a dial, not a replay

A connection is a socket; the pool's whole recovery story is to notice a latched failure and dial again. It does not replay topology or re-subscribe. It could not do so honestly: the broker cannot restore an exclusive or auto-delete queue, and a consumer re-established across a gap has no way to know what was delivered into it. A silent replay would hide the gap rather than close it, so re-establishing what an application declared is the application's, from onConnect, where it can also decide what to do about the messages it may have missed.

The one thing the pool must get right is the count: a dead connection is dropped before a replacement is dialled, so size bounds how many sockets are open even under concurrent borrows.

A session re-runs setup; it does not replay

The pool lends connections for a scoped piece of work. A long-lived consumer needs more: something that outlives the socket. A Session is that, and it recovers the way Node's amqp-connection-manager does rather than the way the Java client does. The Java client replays a topology cache it kept as the application declared things; its own guide notes the cache cannot see a queue that expired or auto-deleted, and replay cannot restore an exclusive queue honestly. A session instead re-runs the setup function each recovering channel was given, so what is declared after a reconnect is exactly what the application says, every time.

Where both of those clients carry legacy compromises, the session does not. Retries back off exponentially with jitter instead of on a fixed interval, so a restarted cluster's clients do not all return in the same instant. The first connection is retried too while its failures are transient (the Java client does not retry it), but refused credentials fail at once. A publish made while disconnected waits, within its timeout, instead of being lost (Java) or queued in memory without bound (amqp-connection-manager). And a publish whose confirm was outstanding at the drop rejects instead of being republished, since its fate at the broker is unknown and a republish could deliver it twice.

Deliveries are where recovery meets the protocol. A delivery tag belongs to its channel, tags restart on a new one, and the broker requeued whatever was unacknowledged when the old channel died. So settling a delivery from before the drop sends nothing — a stale tag on the new channel would be a protocol error, or worse, acknowledge a different message — and one that was buffered but never read is not handed over, because the broker delivers it again. Processing is therefore at-least-once, as it is for any recovering consumer. (DECISIONS D177.)

What it costs

The client is JavaScript, so the wire codec — the decoder's buffer, each frame, each reassembled message — lives on the JavaScript heap. That is the same line runtime:db draws: Rust owns transport and the embedded engine, and JavaScript owns every network protocol, so a networked backend adds no runtime code. The cost is memory and the collector's time over it, not a capability — nothing here needs a grant beyond the socket it already had. Moving the codec into Rust later would not change what this package exports.

Not here

AMQP 1.0 and the RabbitMQ Streams protocol, the management HTTP API, the tx (transactions) class, the access class, and basic.recover-async.