[Bug] Data loss in gRPC stream push: dead stream is never deregistered and undelivered messages get ACKed

### Search before asking

- [x] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) and found no similar issues.

### Environment

Linux

### EventMesh version

Other

### What happened

### Environment

- OS: Linux
- JDK: 8
- Deployment: standalone EventMesh runtime, subscriber attached via the gRPC **stream** subscription (`subscribeStream`)

### EventMesh version
1.12.0

### What happened

During a sustained high-throughput consumption run (tens of millions of messages) delivered to a subscriber over the gRPC stream subscription, a large fraction of messages was **silently lost** after the subscriber's gRPC stream broke mid-run.

Symptoms:

- The subscriber's application-level **heartbeat kept succeeding**, but its gRPC push **stream was already closed**, and the client did not re-subscribe.
- The runtime kept pushing to the now-dead stream, producing a storm of `Stream is already completed` errors (hundreds of thousands of occurrences), pinning one CPU core at 100%.
- The failed pushes went through the retry path and were ultimately **acknowledged (offset advanced via `updateOffset`) rather than redelivered**. The broker then considered those messages consumed, so they were never redelivered — permanent loss.

Net effect: under a *single* stream disconnect, at-most-once delivery degraded into large-scale, unrecoverable data loss.

### Root cause

We traced it to four contributing defects in the gRPC stream push path. Class/method names are from the version we examined and may differ slightly on current master:

1. **A dead stream is never deregistered.** `ConsumerService#subscribeStream`'s `onError` / `onCompleted` only close the stream; they never remove the client's `EventEmitter` from the topic's emitter set. The dead emitter lingers and keeps getting selected for push.

2. **`StreamPushRequest` snapshots the emitter list at construction time.** When clients (re-)register, the emitter list is rebuilt, but already-created in-flight / retry push requests keep a **stale reference** and forever target the old (dead) emitter.

3. **Liveness detection only looks at the heartbeat, not the stream.** Session-expiry cannot evict a client whose heartbeat is alive but whose push stream is dead — exactly this failure mode.

4. **Failure paths ACK undelivered messages.** When retries are exhausted (`AbstractPushRequest#delayRetry` → `complete()`), or a "no available emitter" / "already handled" branch is taken, the request ends via `complete()`/`finish()` → `updateOffset` — i.e. it **advances the offset for a message that was never delivered**, instead of negatively acknowledging / redelivering it. No path redelivers a message whose push provably failed.

Combined: once a stream silently dies, every message routed to it is pushed → fails → retried → then ACKed away.

### How to reproduce

1. Subscribe a client via the gRPC **stream** subscription and start a sustained high-throughput publish → consume run.
2. Mid-run, kill the client's push stream **without** unsubscribing (drop the connection / kill the client) while its heartbeat keeps beating — i.e. the stream closes but the session stays registered.
3. Observe the `Stream is already completed` storm, offsets advancing, and the downstream sink missing the messages "pushed" to the dead stream.

### Impact

Silent, unrecoverable data loss on the gRPC stream push path under a common failure (subscriber stream disconnect, including slow-consumer-induced disconnects). Because offsets are advanced, recovery requires an external replay.

### Debug logs

```Java
- Deregister the emitter on `onError` / `onCompleted`, keyed by **emitter identity** (not `ip:pid`, so a freshly reconnected client reusing the same address is not evicted by mistake).
- Resolve the current emitter list **at push time** (from the topic config) inside `StreamPushRequest`, instead of snapshotting at construction; skip cancelled/closed streams.
- Factor **stream liveness** (not just heartbeat) into session expiry.
- On push failure / no-available-emitter / retry exhaustion, do **not** advance the offset — negatively acknowledge / redeliver. More broadly, consider gating offset advance on a confirmed downstream delivery/ACK (at-least-once) and letting a downstream idempotency key absorb duplicates.

We implemented the above and a follow-up fault-injection run (kill the subscriber mid-consumption) then showed **zero loss at the runtime layer**. Happy to contribute a PR.
```

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) *

Start with ConsumerService#subscribeStream and trace its onError/onCompleted cleanup into the topic emitter set. Then inspect StreamPushRequest and AbstractPushRequest#delayRetry, including complete()/finish() and updateOffset, to understand stale emitters and failure acknowledgements. Done means dead streams are removed, current emitters are resolved at push time, stream liveness is considered, failed deliveries are redelivered, and the described fault-injection run shows zero runtime-layer loss.

Written by the indexing model from the issue text.