[Bug] Server graph metadata can permanently diverge from PD when graph watch events are missed

### Bug Type

server / metadata synchronization / distributed consistency

### Summary

In distributed mode with PD enabled, PD graph metadata is the durable source of truth, while each HugeGraph Server maintains a local derived state:

- local `GraphManager` graph instances
- embedded Gremlin Server graph bindings
- traversal source bindings such as `__g_-`

Currently, Server relies primarily on PD KV watch events such as `GRAPH/ADD` to keep this local state synchronized.

However, the current watch mechanism is not replayable, and graph event handling has no reconciliation fallback.

As a result, a Server can permanently miss a graph that already exists in PD.

This is related to #3137 and complements #3138.

#3138 guarantees that the **Server handling CreateGraph** has completed its own local graph/Gremlin registration before publishing the graph to PD.

The remaining problem is that **other Server replicas are still converging asynchronously from PD metadata**, and there is currently no mechanism guaranteeing that a replica eventually catches up if an incremental notification is lost or local processing fails.

---

### Current synchronization flow

```mermaid
flowchart TD
A[Create graph on Server A] --> B[Server A opens graph]
B --> C[Server A registers local Gremlin binding]
C --> D[Persist GRAPH_CONF in PD]
D --> E[Write GRAPH/ADD event]

E --> F[Server B PD watch]
F --> G[graphAddHandler]
G --> H[Read graph config from PD]
H --> I[Create local graph]
I --> J[Register Gremlin binding]

G -. callback failure .-> X[Event lost]
E -. watch disconnected .-> Y[Event missed]
X --> Z[Server B permanently lacks graph]
Y --> Z
```

The desired invariant should be:

```text
If a graph exists in durable PD graph metadata,
every eligible Server should eventually have the corresponding
local graph instance and Gremlin binding.
```

Currently that invariant is not guaranteed.

---

### Confirmed failure modes

#### 1. Graph event handler failure is not retried

`GraphManager.graphAddHandler()` loads the graph from PD and calls:

```java
createGraph(graphSpace, graphName, creator, config, false)
```

If graph construction fails, the exception is eventually caught by `ConsumerWrapper`, which only logs:

```java
LOG.error("Listener exception occurred.", e);
```

There is no retry, nack, or later reconciliation.

Therefore a transient local error can result in:

```text
PD contains graph A
|
GRAPH/ADD delivered to Server B
|
Server B graph load fails once
|
event processing ends
|
Server B never loads graph A
```

---

#### 2. PD KV watch has no event replay

`WatchRequest` currently contains:

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

There is no revision, sequence, offset, or last-consumed event ID.

On the PD side, watch events are pushed only to observers that are online at the time of the KV mutation.

Therefore events produced while a Server is disconnected cannot be replayed later.

```mermaid
sequenceDiagram
participant S as Server B
participant P as PD

S->>P: watch GRAPH/ADD
P-->>S: watch established

P--xS: connection lost

Note over P: create graph A
Note over P: create graph B

S->>P: reconnect watch
P-->>S: watch established again

Note over S: graph A/B events are not replayed
```

---

#### 3. Startup has a snapshot-to-watch gap

Current startup order in `GraphManager.loadMetaFromPD()` is effectively:

```java
loadGraphsFromMeta(graphConfigs());
listenMetaChanges();
```

This creates a race:

```mermaid
sequenceDiagram
participant S as Starting Server
participant P as PD
participant O as Other Server

S->>P: scan graph configs
P-->>S: A, B

O->>P: create graph C
Note over P: GRAPH_CONF/C persisted
Note over P: GRAPH/ADD/C emitted

S->>P: register GRAPH/ADD watch

Note over S: local state = A, B
Note over P: desired state = A, B, C
```

Graph C is neither included in the original snapshot nor received by the later watch.

Since the watch does not support replay, Server remains inconsistent until restart.

Simply changing the order to "watch first, then scan" is also not a complete solution because watch registration itself is asynchronous and there is no durable revision boundary between the snapshot and incremental stream.

---

### Relationship with #3137 and #3138

#3137 identified two related distributed graph creation problems:

1. the creating Server could return before its own Gremlin binding existed;
2. other Server replicas converge asynchronously and have no cluster-wide readiness guarantee.

#3138 fixed the first problem by making local graph registration synchronous before publishing graph metadata.

This issue focuses on a different but related correctness property:

> Even if cluster-wide immediate readiness remains asynchronous, every Server must eventually converge to PD state.

```mermaid
flowchart LR
A[#3137] --> B[#3138]
A --> C[This issue]

B --> D[Creator Server locally ready before publish]
C --> E[Remote Servers eventually converge]

D --> F[Improved graph lifecycle correctness]
E --> F
```

This issue does **not** attempt to guarantee that every Server is ready immediately when CreateGraph returns.

That stronger cluster-wide readiness guarantee can remain a separate follow-up.

---

### Proposed solution

Keep PD watch as the fast path, and add a lightweight desired-state reconciliation path based on durable graph configs.

```mermaid
flowchart TD
P[PD durable GRAPH_CONF] --> W[Graph-add watch]
P --> R[Low-frequency reconciliation]

W --> L[Load graph locally]
R --> D{PD graph exists locally?}

D -->|yes| N[No action]
D -->|no| L

L --> G[Create graph instance]
G --> B[Register Gremlin binding]
```

The reconciliation only needs to handle:

```text
PD has graph
AND
local Server does not have graph
->
load graph from PD
```

For the first implementation, it does not need to automatically drop graphs that exist locally but no longer exist in PD. The existing `GRAPH/REMOVE` path can continue handling removals.

This keeps the scope small and avoids introducing a full controller/state-machine implementation.

A possible implementation is:

```java
for (Map.Entry> graph : graphConfigs().entrySet()) {
if (!graphs.containsKey(graphName) &&
!creatingGraphs.contains(graphName) &&
!removingGraphs.contains(graphName)) {
loadGraphFromMetaIfAbsent(...);
}
}
```

The existing graph-add handler and reconciler should reuse the same idempotent graph loading helper.

---

### Why reconciliation instead of Gremlin request fallback

Another possible mitigation is to query PD when a Gremlin request references a missing graph.

However, that moves metadata consistency logic into the request path and introduces additional concerns:

- repeated lookups for truly nonexistent graphs
- negative caching
- request-side rate limiting
- concurrent load deduplication
- direct Gremlin connections bypassing the REST fallback

The synchronization layer itself should guarantee eventual convergence instead.

The request path should not be responsible for repairing control-plane state.

---

### Acceptance criteria

- A graph that exists in PD but is absent on one Server is automatically loaded without restarting that Server.
- A transient failure in graph-add event processing does not permanently lose the graph.
- A graph created during a PD watch outage is eventually loaded after connectivity is restored.
- Existing normal graph-add watch behavior remains the low-latency path.
- Reconciliation does not recreate a graph currently being removed by the same Server.
- No Gremlin request-path PD lookup is introduced.

---

### Work status

I am currently working on this and plan to submit a focused PR implementing graph desired-state reconciliation and the corresponding regression tests.

Related PD watch reliability issue: #3152

Start by reading GraphManager.loadMetaFromPD(), listenMetaChanges(), and graphAddHandler(), then inspect WatchRequest and the existing graph-add path. Trace how graph configs are loaded from PD and how local GraphManager and Gremlin bindings are registered. Done means durable graphs are eventually loaded after missed events or transient failures without recreating graphs being removed, with regression tests covering the listed acceptance criteria.

Written by the indexing model from the issue text.