[Bug] Shared dispatcher leaks negative aggregate availablePermits on consumer removal: deferred flow credit is discarded while the full consumer counter is debited

### Search before reporting

- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Read release policy

- [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

### User environment

Reproduction environment (where the defect is demonstrated):

- Broker: current `master` @ 00a6badafc, in-process broker test (`pulsar-broker` test infrastructure), Gradle 9.7
- Java: OpenJDK 21 (Temurin), Linux x86_64
- Client library: official Java client (in-tree test)

Original production incident (what led us to read the broker source):

- Persistent topic, `Shared` subscription, non-partitioned; consumers behind a third-party client whose wire behaviour we ruled out first (the binary protocol carries only monotonic client→broker permit increments, so no client can express a permit decrement)
- Production broker version available on request; the accounting shown below is unchanged on current master

### Issue Description

**What happened.** A `Shared` subscription permanently stopped dispatching after a consumer-churn window (a cursor reset performed with consumers still attached, a 12→1 consumer scale-down, then an instance recycle mid-drain). The broker reported `availablePermits` at **-177300** while `msgBacklog` was non-empty, `unackedMessages` accounting was clean, and the connection stayed healthy. Only `pulsar-admin topics unload` recovered the subscription.

**What we found in the broker source** (all references on `master` @ 00a6badafc). The Shared dispatcher's per-subscription aggregate can be debited permits that were never credited to it:

| Site | Location | Role |
| --- | --- | --- |
| Consumer counter credit | `Consumer.java:919` (`MESSAGE_PERMITS_UPDATER.getAndAdd`) | synchronous, IO thread |
| Aggregate credit deferral | `PersistentDispatcherMultipleConsumers.java:302` (`consumerFlow` → broker executor) | asynchronous |
| Aggregate credit **discard** | `PersistentDispatcherMultipleConsumers.java:309` (`if (!consumerSet.contains(consumer)) return;`) | drops the credit if the consumer left in between |
| Aggregate debit on removal | `PersistentDispatcherMultipleConsumers.java:261` (`totalAvailablePermits -= consumer.getAvailablePermits()`) | debits the **full** consumer counter, discarded credit included |
| Only reset | `PersistentDispatcherMultipleConsumers.java:298` (last consumer out) | never reached while ≥ 1 consumer remains |

So when a consumer's `CommandFlow` races its own removal, the per-consumer counter is credited (synchronously) but the aggregate credit is discarded — and removal then debits the aggregate by the full counter. The loss is unbounded across churn rounds and survives any scale-down that keeps at least one consumer attached, which matches our 12→1 incident. It is a **subtract-without-credit**, not a double subtraction: the `consumerSet.removeAll(consumer) == 1` guard at `:244` (from #22270) already prevents double removal from debiting twice.

**Two related observations:**

1. Since #10417 (`3550f2e7c1`), `readMoreEntries` takes `Math.max(totalAvailablePermits, firstAvailableConsumerPermits)` (`:366`), so a negative *aggregate* alone no longer stalls dispatch on master. The production wedge is most consistent with the **per-consumer** path: `Consumer.sendMessages` (`Consumer.java:434`) can drive a consumer's counter negative (`getMaxEntriesInThisBatch` forces ≥ 1 entry; a non-writable consumer substitutes `availablePermits = 1` at `:839`), `isConsumerAvailable` requires `> 0`, and Flow is monotonic — a deeply negative consumer is unrecoverable from the client side. Admin-exposed `availablePermits` is `ConsumerStats`-level, matching the figure we observed.
2. Secondary defect in the same method: `addUnAckedMessages(-consumer.getUnackedMessages())` at `:243` sits **outside** the `consumerSet` guard, so removing an already-removed consumer double-decrements `totalUnackedMessages`.

**Why we believe this is a bug.** The subscription aggregate should be a conserved quantity — debited exactly what was credited for each consumer. The discarded-credit/full-debit asymmetry violates that invariant deterministically, as the in-tree test below shows.

### Error messages

```text
testRemovingConsumerDoesNotDebitPermitsThatWereNeverCredited FAILED
java.lang.AssertionError: subscription aggregate availablePermits went negative after
removing a consumer whose in-flight flow credit was discarded:
before=10, flow=1010, after=-1005
```

(Arithmetic is self-verifying: before = 2 consumers × receiverQueueSize 5; the departing consumer holds 5 + 1010 = 1015; 10 − 1015 = −1005.)

### Reproducing the issue

Branch with the tests: https://github.com/FlorentinDUBOIS/pulsar/tree/repro/shared-permits-negative-aggregate (`pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SharedSubscriptionAvailablePermitsInvariantTest.java`)

```bash
./gradlew :pulsar-broker:test -PtestRetryCount=0 \
--tests "org.apache.pulsar.broker.service.persistent.SharedSubscriptionAvailablePermitsInvariantTest"
```

- `testRemovingConsumerDoesNotDebitPermitsThatWereNeverCredited` — deterministic probe: fails on master, identical value across repeated runs. Known limitation, documented in-code: it forces the interleaving by holding the dispatcher monitor (the monitor both `removeConsumer` and `internalConsumerFlow` synchronize on), standing in for a busy broker executor.
- `testSubscriptionAvailablePermitsNeverNegativeUnderConsumerChurn` — real Java clients against the in-process broker, 4 seeded churn rounds mixing abrupt client shutdowns with unacked messages in flight, connection drops, graceful closes, `resetCursor` with consumers attached, and re-subscribes; asserts the aggregate and every per-consumer counter stay ≥ 0 and that survivors drain a fresh batch. It **passes** on master — those patterns did not fire the race in-process over our runs — and stands as a guard for the invariant.

We had also reproduced the exact production signature (wedge, healthy connection, only unload recovers, recovery credits exactly one receiver-queue window per in-place re-subscribe) in a deterministic client-side simulation before reading the broker source; the simulation predicted the accounting shape this report demonstrates.

### Additional information

- Related, same symptom family: #24418 (our earlier report of negative per-consumer `availablePermits`, closed after a proxy reboot cleared the visible symptom — the accounting asymmetry reported here is independent of any proxy), #10813, #24926.
- Questions for maintainers: (1) do you confirm the ordering-asymmetry reading above? (2) preferred fix shape — debit only the credited portion / recompute the aggregate from remaining consumers on removal, versus making the credit non-discardable? (3) should a negative aggregate or consumer counter be surfaced by a metric/alarm? It is silent today.
- We have a fix in progress on the same branch (both defects, with regression tests) and are happy to open it as a PR.

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Run the named Gradle test and read SharedSubscriptionAvailablePermitsInvariantTest.java first. Trace Consumer.java:919 and :434, then PersistentDispatcherMultipleConsumers.java:243, :261, :298-309, and :366 to verify the accounting interleaving. Done means regression coverage for both removal defects and nonnegative aggregate and per-consumer permit accounting under churn.

Written by the indexing model from the issue text.