Federation links do not respect resource alarms on the downstream node

[lukebakken]

Problem

Federation exchange and queue links continue publishing messages to the downstream node during disk and memory resource alarms. This can accelerate the condition that triggered the alarm (e.g., filling disk or exhausting memory).

All other publishing paths in RabbitMQ respect resource alarms:

• Network (AMQP 0-9-1) connections receive connection.blocked and cooperatively stop publishing
• AMQP 1.0 sessions have their incoming-window set to 0 via a FLOW frame
• The local shovel registers directly with rabbit_alarm and buffers messages until alarms clear
• The AMQP 0-9-1 shovel registers a blocked handler on the destination connection

Federation links are the exception.

Root Cause

Federation's downstream connection uses #amqp_params_direct{} (an in-process connection to the local node). The alarm notification machinery is present but unused:

1. rabbit_direct:connect/5 registers the connection pid with rabbit_alarm
2. amqp_gen_connection receives {conserve_resources, ...} casts and tracks blocked_by state
3. When blocked, it sends #'connection.blocked'{} to the registered block_handler pid

However, federation never calls amqp_connection:register_blocked_handler/2 on the downstream connection (DConn), so the blocked notification is sent to none and discarded.

Additionally, federation publishes via amqp_channel:cast/3 (the noflow variant), so it does not participate in credit flow backpressure from queues. This means even queue-level backpressure (quorum queue soft-limit {block, QName} actions, classic queue credit flow) does not reach federation. A queue that is overloaded but has not yet triggered a system-wide alarm has no protection from federation publishing.

The server-side rabbit_channel has no alarm check in its handle_method(#'basic.publish'{}, ...) path - it routes and delivers unconditionally. Neither rabbit_queue_type:deliver/4 nor the individual queue type implementations (rabbit_fifo_client:enqueue, classic queue deliver) reject messages based on resource alarms.

Impact

During a resource alarm, federation links continue to deliver messages to local queues at whatever rate the upstream provides them. The upstream basic.qos prefetch (default 1000) bounds in-flight messages at any instant, but since downstream confirms return quickly (queue acceptance is not alarm-gated), the prefetch window refills continuously.

This means federation can sustain full throughput into local queues during alarms, potentially:

• Filling disk further during a disk alarm
• Increasing memory pressure during a memory alarm
• Prolonging or worsening the alarm condition

Even below alarm thresholds, a backed-up queue cannot apply backpressure to federation because the noflow publish path bypasses credit flow entirely.

Affected Modules

• deps/rabbitmq_exchange_federation/src/rabbit_federation_exchange_link.erl
• deps/rabbitmq_queue_federation/src/rabbit_federation_queue_link.erl
• deps/rabbitmq_federation_common/src/rabbit_federation_link_util.erl

Suggested Fix

Two related gaps should be addressed together:

1. Resource alarm blocking

Register a blocked handler on the downstream connection and pause forwarding when blocked:

1. In rabbit_federation_link_util:start_conn_ch/5, after opening the downstream connection, call amqp_connection:register_blocked_handler(DConn, self())
2. In both link modules, handle #'connection.blocked'{} in handle_info - set a blocked flag in state
3. Handle #'connection.unblocked'{} - clear the flag and drain any buffered messages
4. In handle_info({#'basic.deliver'{}, ...}), check the blocked flag before calling forward/9 - if blocked, buffer the message

The buffer is naturally bounded by the upstream prefetch_count (default 1000). Once the link stops acking upstream messages (because it stops forwarding), the upstream channel stops delivering. When the alarm clears and the buffer drains, upstream delivery resumes.

This mirrors the approach used by rabbit_amqp091_shovel, which registers a blocked handler and maintains a pending queue with the same bounded-by-prefetch property.

Note: amqp_gen_connection already handles the race where an alarm is active at registration time - it immediately sends #'connection.blocked'{} to the newly registered handler.

2. Queue-level backpressure via credit flow

Switch from amqp_channel:cast/3 (noflow) to amqp_channel:cast_flow/3 for downstream publishes. This enables credit flow between the federation link process and the downstream channel, allowing the channel to apply backpressure when queues return {block, QName} actions.

The link process would need to handle {bump_credit, ...} messages and track whether it is blocked by credit flow (similar to how rabbit_amqp091_shovel calls credit_flow:handle_bump_msg/1 and checks credit_flow:blocked/0).

This provides queue-level backpressure before conditions escalate to a system-wide alarm.

[Vishal-770]

Hi , I'd like to pick this issue up!

I've been looking through the federation link code, and I agree with the suggested two-part fix. Here is my plan on how to implement it:

1. Resource Alarm Blocking

• In rabbit_federation_link_util:start_conn_ch/5, I'll register the link process using amqp_connection:register_blocked_handler(DConn, self()) right after the downstream connection is established.
• In both rabbit_federation_exchange_link and rabbit_federation_queue_link, I'll update the #state{} to track a blocked flag and a blocked_buffer queue.
• I'll add handle_info clauses to catch #'connection.blocked'{} and #'connection.unblocked'{}. When blocked, incoming #'basic.deliver'{} messages will be buffered (naturally bounded by the upstream prefetch count). When unblocked, the buffer will be drained.

2. Queue-Level Backpressure (Credit Flow)

• I'll update rabbit_federation_link_util:forward/9 to use amqp_channel:cast_flow/3 instead of the noflow cast.
• I will add {bump_credit, Msg} handlers in the link processes to integrate with credit_flow:handle_bump_msg/1 and credit_flow:blocked/0, ensuring we buffer messages if credit flow backpressure is applied before a system-wide alarm is triggered.

I'm starting on the implementation now and should have a PR up shortly. Let me know if you have any early feedback on this approach!

[Vishal-770]

@lukebakken I have implemented the fix for this issue over in PR #17238!

The implementation closely follows the strategy used in the rabbit_amqp091_shovel plugin. I've registered the connection.blocked handler and integrated cast_flow so that federation respects both global resource alarms and queue-level credit flow backpressure, with the buffer naturally bounded by the upstream prefetch limit.

Please let me know if you get a chance to review it or if you'd like me to make any adjustments!

[lukebakken]

I'll do a review in the coming days, thanks.