Block direct shovel publishes during alarms

[the-mikedavis]

Is your feature request related to a problem? Please describe.

Currently direct connections (amqp_direct_connection) do not notice when publishers are blocked due to alarms like disk or memory usage. For an example:

1. mkdir /tmp/small-disk
2. sudo mount -t tmpfs -o size=3000M small-disk /tmp/small-disk
3. TEST_TMPDIR=/tmp/small-disk make run-broker
4. sbin/rabbitmqctl set_disk_free_limit 1000M
5. Publish until the disk alarm is triggered (perf-test -qq -u qq -x 1 -y 0)
6. sbin/rabbitmqctl set_parameter shovel my-shovel '{"src-protocol": "amqp091", "src-uri": "amqp://", "src-queue": "qq", "dest-protocol": "amqp091", "dest-uri": "amqp://", "dest-queue": "qq-copied"}'

The shovel can successfully move the messages from qq to qq-copied despite publishers being blocked. If the disk free limit is set low enough, moving the messages in a queue like this can fully exhaust disk space. It also probably works to shovel in messages from another instance for the same effect. As long as the destination connection is direct, publishing will not be blocked by alarms.

Describe the solution you'd like

Direct connections can register themselves with rabbit_alarm and emit a #'connection.blocked'{} when an alarm fires. rabbit_amqp091_shovel then avoids sending messages to the destination (since #6224).

Describe alternatives you've considered

No response

Additional context

No response

[gomoripeti]

This test case utilises the fact that direct shovels are not blocked during an alarm.

rabbitmq-server/deps/rabbitmq_shovel/test/amqp091_dynamic_SUITE.erl

Line 802 in 2303570

%% so we use a temporary shovel with direct

Probably need to switch that test case to use a local connection instead