fix: preserve traversal and PD client contracts

docs/storage-lifecycle.md

@@ -99,6 +99,14 @@ the interrupt flag; input that never entered the queue cannot be reported as acc
 Ephemeral jobs retain their negative task IDs and existing wraparound rule. Concurrent
 schedulers allocate these IDs atomically, avoiding duplicate IDs caused by lost updates.
 
+Algorithm traversal helpers restore the caller's query capacity and close the traversal
+on success or failure. If calculation and close both fail, the calculation error stays
+primary and the close error is suppressed. String IDs and values in algorithm JSON
+results preserve quotes, backslashes and control characters through JSON encoding;
+numeric result values remain JSON numbers.
+Vertex and edge counting also close their scan iterator when reading or reporting
+progress fails, preserving the calculation error if closing fails as well.
+
 ### Store sort and shuffle iterators
 
 Sort/shuffle applies the requested comparator both in memory and after spilling to
@@ -116,6 +124,14 @@ them is terminal and discards pending values and completion markers. The source
 closed once; a failed close remains visible on later close calls. ORDER BY propagates
 sort failures to the query's existing error and cleanup path with the original cause.
 
+Union and intersection filters preserve a pending record across
+repeated probes, including the extra probe at RPC batch boundaries. Reading with
+`next()` consumes that record exactly once. Union results also support reading without
+a preceding probe. Closing either filter discards pending results and stops further
+probes of its source.
+For a spilled intersection, the requested group size also applies to the final group;
+the default mode continues to select groups containing more than one record.
+
 ### Callback executor configuration
 
 The gRPC application callback queue is always unbounded (`Integer.MAX_VALUE`), matching
@@ -138,6 +154,16 @@ the RPC method and original cause. Existing retry budgets and response-header er
 codes are preserved, so registration and metadata callers receive a PD error rather
 than a missing response.
 
+An interrupted wait for the first lock client's initialization stops before issuing a
+lock RPC, preserves the interrupt flag and leaves the initialization permit with its
+owner. Successful leader discovery binds subsequent PD calls to the discovered leader,
+including when the configured peer used for discovery is a follower.
+
+PD watch notifications isolate runtime failures from individual business listeners.
+The failure is logged and the remaining listeners still receive the event on the same
+callback thread. Cache updates retain their existing order and errors still propagate;
+failed listener work is not retried automatically.
+
 PD snapshot compression finishes the ZIP central directory before flushing and syncing
 the archive. An unreadable source directory fails with an I/O error. Archive entries
 and the checksum format remain compatible with existing snapshots.

hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java

@@ -580,7 +580,7 @@ private void reconnect(WatchSubscription subscription) {
         }
     }
 
-    private void acquire(AtomicLong clientId, Semaphore semaphore) {
+    private void acquire(AtomicLong clientId, Semaphore semaphore) throws PDException {
         if (clientId.get() == 0L) {
             try {
                 semaphore.acquire();
@@ -589,7 +589,9 @@ private void acquire(AtomicLong clientId, Semaphore semaphore) {
                 }
                 log.info("wait for client starting....");
             } catch (InterruptedException e) {
-                log.error("get semaphore with error:", e);
+                Thread.currentThread().interrupt();
+                throw new PDException(Pdpb.ErrorType.UNKNOWN_VALUE,
+                                      "Interrupted while waiting for lock client initialization", e);
             }
         }
     }
@@ -704,8 +706,8 @@ public void close() {
             return;
         }
         reconnectScheduler.shutdownNow();
-        if (reconnectExecutor instanceof ExecutorService) {
-            ((ExecutorService) reconnectExecutor).shutdownNow();
+        if (reconnectExecutor instanceof ExecutorService executor) {
+            executor.shutdownNow();
         }
         release(lockSemaphore);
         for (WatchSubscription subscription : subscriptions) {

hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/PDClient.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 import org.apache.hugegraph.pd.client.impl.PDPulseImpl2;
 import org.apache.hugegraph.pd.client.interceptor.Authentication;
@@ -52,6 +53,7 @@
 import org.apache.hugegraph.pd.grpc.Pdpb.GetPartitionResponse;
 import org.apache.hugegraph.pd.grpc.Pdpb.GraphStatsResponse;
 import org.apache.hugegraph.pd.grpc.watch.WatchResponse;
+import org.apache.hugegraph.pd.grpc.watch.WatchType;
 import org.apache.hugegraph.pd.watch.NodeEvent;
 import org.apache.hugegraph.pd.watch.PartitionEvent;
 import org.apache.hugegraph.pd.watch.PartitionEvent.ChangeType;
@@ -165,7 +167,8 @@ public void onNext(PartitionEvent response) {
                 if (response.getChangeType() == ChangeType.DEL) {
                     cache.removeAll(response.getGraph());
                 }
-                listeners.forEach(listener -> listener.onPartitionChanged(response));
+                notifyListeners(WatchType.WATCH_TYPE_PARTITION_CHANGE,
+                                listener -> listener.onPartitionChanged(response));
             }
 
             @Override
@@ -200,7 +203,8 @@ public void onNext(NodeEvent response) {
                     }
                 }
 
-                listeners.forEach(listener -> listener.onStoreChanged(response));
+                notifyListeners(WatchType.WATCH_TYPE_STORE_NODE_CHANGE,
+                                listener -> listener.onStoreChanged(response));
             }
 
             @Override
@@ -214,7 +218,8 @@ public void onError(Throwable throwable) {
         graphWatcher = pdWatch.watchGraph(new PDWatch.Listener<>() {
             @Override
             public void onNext(WatchResponse response) {
-                listeners.forEach(listener -> listener.onGraphChanged(response));
+                notifyListeners(WatchType.WATCH_TYPE_GRAPH_CHANGE,
+                                listener -> listener.onGraphChanged(response));
             }
 
             @Override
@@ -245,7 +250,8 @@ public void onNext(WatchResponse response) {
                             break;
                     }
                 }
-                listeners.forEach(listener -> listener.onShardGroupChanged(response));
+                notifyListeners(WatchType.WATCH_TYPE_SHARD_GROUP_CHANGE,
+                                listener -> listener.onShardGroupChanged(response));
             }
 
             @Override
@@ -256,6 +262,17 @@ public void onError(Throwable throwable) {
 
     }
 
+    private void notifyListeners(WatchType eventType, Consumer<PDEventListener> action) {
+        for (PDEventListener listener : this.listeners) {
+            try {
+                action.accept(listener);
+            } catch (RuntimeException e) {
+                String listenerClass = listener == null ? "null" : listener.getClass().getName();
+                log.error("PD watch listener {} failed to handle {} event", listenerClass, eventType, e);
+            }
+        }
+    }
+
     private synchronized void closeStub(boolean closeWatcher) {
         // TODO ManagedChannel  Did not close properly
         proxy.set(null);
@@ -314,7 +331,7 @@ private String newLeaderStub() {
                 var leaderIp = getLeaderIp(stub);
                 if (!leaderIp.equalsIgnoreCase(host)) {
                     leaderHost = leaderIp;
-                    proxy.set(getStub(PDGrpc.newBlockingStub(channel)));
+                    proxy.set(getStub(PDGrpc.newBlockingStub(getChannel(leaderIp))));
                 } else {
                     proxy.set(stub);
                     leaderHost = host;

hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/client/KvClientLockTest.java

@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.pd.client;
+
+import java.lang.reflect.Field;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.Pdpb;
+import org.apache.hugegraph.pd.grpc.kv.LockRequest;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
+
+import io.grpc.MethodDescriptor;
+
+@RunWith(Parameterized.class)
+public class KvClientLockTest {
+
+    private final boolean reentrant;
+
+    public KvClientLockTest(boolean reentrant) {
+        this.reentrant = reentrant;
+    }
+
+    @Parameterized.Parameters(name = "reentrant={0}")
+    public static Object[][] lockMethods() {
+        return new Object[][]{{true}, {false}};
+    }
+
+    @Test(timeout = 10000)
+    public void testPreInterruptedLockDoesNotSendRequest() throws Exception {
+        try (ControlledKvClient client = new ControlledKvClient(false)) {
+            AtomicReference<Throwable> failure = new AtomicReference<>();
+            AtomicBoolean interrupted = new AtomicBoolean();
+            Thread caller = new Thread(() -> {
+                Thread.currentThread().interrupt();
+                callAndCapture(client, failure, interrupted);
+            });
+            try {
+                caller.start();
+                caller.join(5000);
+                Assert.assertFalse(caller.isAlive());
+                assertInterrupted(failure.get(), interrupted.get());
+                Assert.assertTrue(client.requests.isEmpty());
+
+                Assert.assertTrue(callLock(client).getSucceed());
+                Assert.assertEquals(0L, client.requests.get(0).getClientId());
+                Assert.assertTrue(callLock(client).getSucceed());
+                Assert.assertEquals(42L, client.requests.get(1).getClientId());
+            } finally {
+                caller.interrupt();
+                caller.join(5000);
+            }
+        }
+    }
+
+    @Test(timeout = 15000)
+    public void testInterruptedWaitDoesNotReleaseAnotherCallersPermit() throws Exception {
+        try (ControlledKvClient client = new ControlledKvClient(true)) {
+            ExecutorService executor = Executors.newSingleThreadExecutor();
+            AtomicReference<Throwable> failure = new AtomicReference<>();
+            AtomicBoolean interrupted = new AtomicBoolean();
+            Thread waiter = new Thread(() -> callAndCapture(client, failure, interrupted));
+            try {
+                Future<LockResponse> first = executor.submit(() -> callLock(client));
+                Assert.assertTrue(client.firstRequest.await(5, TimeUnit.SECONDS));
+                Semaphore semaphore = lockSemaphore(client);
+                waiter.start();
+                awaitQueuedWaiter(semaphore);
+
+                waiter.interrupt();
+                waiter.join(5000);
+                Assert.assertFalse(waiter.isAlive());
+                assertInterrupted(failure.get(), interrupted.get());
+                Assert.assertEquals(1, client.requests.size());
+                Assert.assertEquals("The first caller must still own the permit", 0,
+                                    semaphore.availablePermits());
+
+                client.releaseFirst.countDown();
+                Assert.assertEquals(42L, first.get(5, TimeUnit.SECONDS).getClientId());
+                Assert.assertEquals(1, semaphore.availablePermits());
+                Assert.assertTrue(callLock(client).getSucceed());
+                Assert.assertEquals(42L, client.requests.get(1).getClientId());
+            } finally {
+                client.releaseFirst.countDown();
+                waiter.interrupt();
+                waiter.join(5000);
+                executor.shutdownNow();
+                Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+            }
+        }
+    }
+
+    private LockResponse callLock(KvClient<?> client) throws PDException {
+        return this.reentrant ? client.lock("key", 1000L) :
+               client.lockWithoutReentrant("key", 1000L);
+    }
+
+    private void callAndCapture(KvClient<?> client, AtomicReference<Throwable> failure,
+                                AtomicBoolean interrupted) {
+        try {
+            callLock(client);
+        } catch (Throwable e) {
+            failure.set(e);
+            interrupted.set(Thread.currentThread().isInterrupted());
+        }
+    }
+
+    private static void assertInterrupted(Throwable failure, boolean interrupted) {
+        Assert.assertTrue(failure instanceof PDException);
+        Assert.assertEquals(Pdpb.ErrorType.UNKNOWN_VALUE, ((PDException) failure).getErrorCode());
+        Assert.assertTrue(failure.getCause() instanceof InterruptedException);
+        Assert.assertTrue(interrupted);
+    }
+
+    private static Semaphore lockSemaphore(KvClient<?> client) throws Exception {
+        Field field = KvClient.class.getDeclaredField("lockSemaphore");
+        field.setAccessible(true);
+        return (Semaphore) field.get(client);
+    }
+
+    private static void awaitQueuedWaiter(Semaphore semaphore) throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (!semaphore.hasQueuedThreads() && System.nanoTime() < deadline) {
+            Thread.sleep(1);
+        }
+        Assert.assertTrue("The second caller must be waiting for initialization",
+                          semaphore.hasQueuedThreads());
+    }
+
+    private static class ControlledKvClient extends KvClient<WatchResponse> {
+
+        private final boolean holdFirst;
+        private final CountDownLatch firstRequest = new CountDownLatch(1);
+        private final CountDownLatch releaseFirst = new CountDownLatch(1);
+        private final List<LockRequest> requests = new CopyOnWriteArrayList<>();
+
+        ControlledKvClient(boolean holdFirst) {
+            super(PDConfig.of("unused:8686"), Mockito.mock(ScheduledExecutorService.class));
+            this.holdFirst = holdFirst;
+        }
+
+        @SuppressWarnings("unchecked")
+        @Override
+        protected <ReqT, RespT> RespT blockingUnaryCall(
+                MethodDescriptor<ReqT, RespT> method, ReqT request) throws PDException {
+            this.requests.add((LockRequest) request);
+            if (this.holdFirst && this.requests.size() == 1) {
+                this.firstRequest.countDown();
+                try {
+                    Assert.assertTrue(this.releaseFirst.await(10, TimeUnit.SECONDS));
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new PDException(Pdpb.ErrorType.UNKNOWN_VALUE, e);
+                }
+            }
+            return (RespT) LockResponse.newBuilder().setHeader(AbstractClient.okHeader)
+                                      .setClientId(42L).setSucceed(true).build();
+        }
+    }
+}