[Bug] PD KvClient watch can permanently stop after reconnect failure

### Bug Type

PD client / reliability / metadata synchronization

### Summary

`KvClient` automatically attempts to reconnect PD KV watches after a gRPC error or PD leader change.

However, the current recovery path performs only one reconnect attempt.

If that attempt also fails, the watch permanently stops.

`onCompleted()` currently has no recovery behavior either.

Since `PdMetaDriver.listen()` and `listenPrefix()` both depend on `KvClient`, this affects all metadata consumers using PD KV watches, including:

- graph lifecycle events
- graph space events
- service metadata events
- cache invalidation events
- other MetaManager listeners

This issue is related to #3137 and to the Server graph reconciliation issue: improving reconnect reliability reduces missed events, while reconciliation is still required because the current watch protocol has no replay mechanism.

---

### Current behavior

`KvClient` handles a watch error approximately as follows:

```mermaid
flowchart TD
A[Watch running] --> B[gRPC error / leader changed]
B --> C[clientId = 0]
C --> D[Call listen again]

D -->|success| E[Watch restored]
D -->|PDException| F[Log warning]
F --> G[Sleep 1 second]
G --> H[Return]
H --> I[Watch permanently stopped]
```

The relevant wrapper currently behaves like:

```java
try {
listen(key, consumer);
} catch (PDException e) {
log.warn(...);
Thread.sleep(1000);
}
```

There is no later retry after the sleep.

Similarly:

```java
@Override
public void onCompleted() {
}
```

does not recreate the watch.

---

### Failure example

```mermaid
sequenceDiagram
participant S as Server
participant C as KvClient
participant P as PD

S->>C: listen(metadata key)
C->>P: open watch
P-->>C: Started

P--xC: connection lost

C->>P: reconnect attempt
P--xC: PD still unavailable

C->>C: log + sleep(1s)
Note over C: reconnect logic exits

Note over S,C: this metadata watch is now permanently gone
```

A temporary PD outage can therefore produce a long-lived Server that appears healthy but no longer receives future metadata changes.

---

### Missed events cannot currently be replayed

Fixing reconnect is important, but reconnect alone is not sufficient to guarantee metadata convergence.

The current protocol contains no revision/offset:

```protobuf
message WatchRequest {
WatchState state = 2;
string key = 3;
int64 clientId = 4;
}
```

The PD service registers the current observer and pushes events directly when KV mutations happen.

There is no persisted event stream used to replay events emitted during disconnection.

```mermaid
sequenceDiagram
participant C as KvClient
participant P as PD

C->>P: watch key
P-->>C: Started

P--xC: disconnected

Note over P: event A
Note over P: event B

C->>P: reconnect
P-->>C: Started

Note over C: A and B are not replayed
```

Therefore this issue should remain focused on **watch connection reliability**, while stateful consumers should use reconciliation against durable PD state when correctness requires recovery of missed events.

---

### Relationship with graph metadata synchronization

Graph lifecycle synchronization currently uses:

```text
durable state:
GRAPH_CONF/

incremental notification:
EVENT/GRAPH/ADD
```

The event key is not an append-only log. Each new graph-add notification overwrites the same event key with the latest graph name.

Therefore missed graph-add events cannot be reconstructed from the event key itself.

```mermaid
flowchart LR
A[PD GRAPH_CONF] -->|durable source of truth| R[Graph reconciliation]
B[PD KV watch] -->|low-latency notification| S[Server]

B -. reconnect reliability .-> C[This issue]
A -. eventual convergence .-> D[Server graph reconciliation issue]

C --> S
D --> S
```

The two fixes are complementary:

- this issue: keep future watch delivery alive after transient failures;
- graph reconciliation: recover durable state changes missed while the watch was unavailable.

---

### Proposed solution

Keep the initial scope small:

1. Continue reconnecting after `onError()` until success or client close.
2. Continue reconnecting after `Leader_Changed`.
3. Recover from unexpected `onCompleted()`.
4. Add bounded retry delay/backoff to avoid tight reconnect loops.
5. Ensure only one reconnect loop exists for each watch.
6. Stop reconnecting cleanly when `KvClient.close()` is called.
7. Add deterministic tests for repeated failures followed by successful recovery.

Example target behavior:

```mermaid
flowchart TD
A[Watch disconnected] --> B{Client closed?}

B -->|yes| C[Stop]
B -->|no| D[Reconnect]

D -->|success| E[Resume watch]
D -->|failure| F[Backoff]
F --> B
```

---

### Out of scope

This issue does not propose:

- adding a durable watch event log;
- adding revisions/offsets to the watch protocol;
- guaranteeing replay of events emitted during downtime;
- changing Server request handling.

Those would be larger architectural changes.

For graph metadata, durable graph configs can already serve as the source of truth, so Server-side reconciliation is a smaller way to guarantee eventual convergence.

---

### Acceptance criteria

- One failed reconnect attempt does not permanently terminate a watch.
- Multiple consecutive connection failures are retried and eventually recover when PD becomes available.
- PD leader changes restore existing watches.
- Unexpected stream completion triggers recovery.
- `close()` prevents further reconnect attempts.
- Reconnect behavior has deterministic test coverage.
- No claim is made that events emitted during the disconnected window are replayed.

---

### Work status

I am currently working on this and plan to submit a focused PR for the reconnect lifecycle and regression tests.

Related Server graph reconciliation issue: #3151

Start with KvClient's onError(), onCompleted(), reconnect path, and close() behavior, then trace how PdMetaDriver.listen() and listenPrefix() depend on it. Add deterministic coverage for repeated reconnect failures, recovery, leader changes, completion, and close stopping retries; done means all listed acceptance criteria pass without claiming event replay.

Written by the indexing model from the issue text.