RabbitMQ pubsub: consumers leak on broker reconnect, parking prefetchCount messages per leaked consumer

## Expected Behavior

After a RabbitMQ broker restart, each subscription should end up with exactly one registered AMQP consumer on its queue, and `prefetchCount` should bound a queue's unacknowledged messages to `prefetchCount × number of subscriber instances`.

## Actual Behavior

A broker restart while messages are in flight leaves **several consumers registered per subscription**, all created by the same sidecar for the same declarative subscription. Only the most recent one delivers to the app; each earlier one stays registered and holds `prefetchCount` messages unacknowledged indefinitely.

1. The broker connection drops. Every subscription sharing the connection sees `listenMessages` fail and enters the reconnect loop in `subscribeForever`.
2. `channel.Cancel(consumerTag)` is attempted, but the channel is already gone, so the deregistration never reaches the broker: `failed to cancel consumer : Exception (504) Reason: "channel/connection is not open"`.
3. One subscriber wins `reconnect()` and creates a new shared channel; the others log `stale reconnect attempt` and loop.
4. While siblings re-declare on the new shared channel, `ensureSubscription` / `channel.Consume` fail intermittently with `Exception (503) Reason: "unexpected command received"`. Each such iteration `break`s **before** reaching the cancel, then loops and calls `channel.Consume` again with a fresh consumer tag.
5. Since the tag is a new UUID per attempt (#4326), the re-registration no longer collides with the previous one, so nothing surfaces the duplicate. The earlier registration survives on a channel that is still alive, holding its prefetch allowance.

Measured with one sidecar, one declarative subscription, `prefetchCount: 100`, and an app that returns 204 promptly:

| after | consumers on queue | messages unacknowledged |
|---|---|---|
| steady state | 1 | 1 |
| 1st broker restart | 2 | 101 |
| 2nd broker restart | 4 | 301 |

Unacknowledged is exactly `(consumers - 1) × prefetchCount + in-flight`. The parked messages never move, while the live consumer keeps draining new traffic normally — so throughput looks healthy and the loss is silent.

With `ttlInSeconds` and `enableDeadLetter` set, the parked messages become permanent loss:

1. After RabbitMQ's `consumer_timeout` (default 30 min) the broker raises `precondition_failed` on the channel and requeues everything unacknowledged.
2. Every requeued message is already past its TTL, so the whole batch is dead-lettered instead of redelivered.

We hit this in a live cluster after a single broker restart: a whole batch of messages was dead-lettered rather than redelivered.

Relevant daprd output (`--log-level debug`), one queue, single subscription:

```
failed to cancel consumer webhook-worker-incoming-webhook-2f7b12e8-...: Exception (504) Reason: "channel/connection is not open"
error in subscriber for webhook-worker-incoming-webhook in ensureSubscription: Exception (504) Reason: "channel/connection is not open"
subscriber is reconnecting in 3s ...
connectionCount: current=3 reference=3
connected with connectionCount=4
setting prefetch count to 100
connectionCount: current=4 reference=3
stale reconnect attempt
error in subscriber for webhook-worker-incoming-webhook in ensureSubscription: Exception (503) Reason: "unexpected command received"
setting prefetch count to 100
registered consumer webhook-worker-incoming-webhook-8ab7c693-... for queue webhook-worker-incoming-webhook
```

Observed on daprd 1.17.7 (components-contrib v1.17.6).

## Steps to Reproduce the Problem

1. Run a single daprd with a RabbitMQ pubsub component using `prefetchCount: 100`, `enableDeadLetter: "true"`, `ttlInSeconds: "60"`, `concurrencyMode: parallel`, `requeueInFailure: "true"`, `autoAck: "false"`, `durable: "true"`, `deliveryMode: "2"`, `publisherConfirm: "true"`.
2. Declare one `dapr.io/v2alpha1` Subscription on a topic, routed to an app that returns `[]` for `GET /dapr/subscribe` and `204` for the topic route after a short sleep, so a backlog builds.
3. Publish continuously:
`while :; do curl -s -o /dev/null -XPOST -H 'Content-Type: application/json' -d '{}' http://localhost:3500/v1.0/publish//; done`
4. Restart the broker container.
5. After ~20s, inspect:
`rabbitmqctl list_queues name messages messages_unacknowledged consumers`
`rabbitmqctl list_consumers queue_name prefetch_count`
6. Repeat step 4 a few times — it fires on roughly one restart in two or three. Adding more subscriptions to the same sidecar makes it fire more readily; with three subscriptions we reached 13 consumers and 1300 unacknowledged messages on one queue.

A self-contained `docker-compose` reproduction is available if useful.

**Root cause location:**
- `components-contrib/pubsub/rabbitmq/rabbitmq.go`: in `subscribeForever`, `channel.Cancel(consumerTag)` runs only after `listenMessages` returns (v1.17.6 lines 539–577). Any iteration that `break`s earlier — `ensureSubscription` or `channel.Consume` failing — skips the cancel entirely, and the next iteration registers a new consumer under a fresh tag.
- The unique-tag change in #4326 (suggested in #4328 as "use unique consumer tags per goroutine") removed the `NOT_ALLOWED - attempt to reuse consumer tag` error that previously made a duplicate registration fail loudly. The duplicate is now silent.

**Possible fixes:**
- Track the live consumer tag per subscription and cancel it against the *current* channel before calling `channel.Consume` again.
- Cancel on every exit path from the inner loop, not only after `listenMessages` returns.
- On a successful `reconnect()`, deregister consumers left over from the previous connection generation for that queue.

## Release Note

RELEASE NOTE: **FIX** RabbitMQ pubsub leaking AMQP consumers on broker reconnect, parking `prefetchCount` messages per leaked consumer.

Read components-contrib/pubsub/rabbitmq/rabbitmq.go, starting at subscribeForever and tracing ensureSubscription, channel.Consume, and reconnect failure paths. Run the self-contained Docker Compose reproduction and inspect rabbitmqctl list_consumers and list_queues; done means one consumer per subscription after broker restarts and unacknowledged messages remain bounded by prefetchCount × subscriber instances.

Written by the indexing model from the issue text.