fix(pd): distinguish maintenance task results from follower no-ops

helm/hugegraph/docs/operations.md

@@ -245,12 +245,28 @@ curl --user "hg:${PD_SECRET}" "${PD_REST}/v1/task/balanceLeaders"
 ```
 
 A busy task returns body `status: 1001`; inspect the result and wait for current
-maintenance to finish rather than looping patrol/balance calls. A follower can
-return an empty success, and leader transfers are asynchronous: HTTP 200 or a
+maintenance to finish rather than looping patrol/balance calls. Images containing
+the task response fix for #3231 return body `status: 100` on a follower, naming the
+known leader's Raft address (`null` while no leader is known). Legacy images can
+still return empty success from followers.
+
+On updated images, partition patrol retains `status` and `partitions` and adds
+`reallocatedShardGroups` (groups whose shard allocation changed) and
+`offlineShardRequests` (group ID to retired Store ID). The latter records requests,
+not successful repairs: inspect group membership to verify their outcome.
+An empty result means patrol ran without scheduling changes. Leader and partition
+balances retain their transfer maps; a partition balance skipped because of its
+maintenance gates returns `{"status":0,"outcome":"skipped"}` instead of an empty
+body. Nonzero `status` identifies a refusal rather than a successful empty plan.
+
+Leader transfers are asynchronous: HTTP 200 or a
 returned transfer plan does not prove completion. Recheck `/v1/partitions` and
 `/v1/stores` for actual leaders, full membership and fresh heartbeats before
 continuing maintenance.
 
+<!-- TODO: simplify the legacy follower-response workaround once supported images
+     include the task response fix for apache/hugegraph#3231. -->
+
 Preserve volumes and recorded member identities during recovery. Deleting a
 RocksDB lock file or assigning an empty volume does not recover shard data.
 The chart does not provide backup or automatic disk-loss recovery; keep backups

hugegraph-pd/docs/api-reference.md

@@ -842,6 +842,28 @@ Point Kubernetes readiness probes, `depends_on` healthchecks and any "wait for P
 
 Match on the body rather than on the status code alone. A PD that predates this endpoint does not reliably answer `404` for it: `RestAuthentication` refuses a request it does not exclude by writing an error envelope, and as of 1.7.0 it does so without setting a status, so an unknown path answers `200` with `{"status":-1,"error":"Unauthorized!"}`. A status-only probe therefore reads such a PD as ready. The body match holds whichever status a refusal carries: a shell gate should use `curl -fsS http://<pd-host>:8620/v1/ready | grep -q '"ready":true'`, and a Kubernetes `httpGet` probe should be paired with a PD image that carries the endpoint.
 
+### Manual maintenance task results
+
+Task endpoints under `/v1/task` require the local PD to be leader. A follower
+answers an error envelope with `status: 100` and the known leader's Raft address
+in `error` (`null` if no leader is known). Check `status`, not HTTP 200 alone.
+`/v1/balanceLeaders` applies the same check. A leadership change during execution
+can also return this error; inspect cluster state before retrying.
+
+`/v1/task/patrolPartitions` retains `status` and `partitions`, and adds
+`reallocatedShardGroups` (IDs whose allocation changed) and `offlineShardRequests`
+(group ID to retired Store ID). Offline-shard requests do not prove completed
+repairs. Empty collections distinguish a completed patrol with nothing to do.
+The changed-group IDs include tombstone-triggered reallocations and are deduplicated
+against the initial shard-count pass.
+
+Leader and partition balances retain their transfer maps. A skipped partition
+balance instead returns `status: 0` with `outcome: skipped`; a completed balance
+with no transfers returns `{}`. Transfers are asynchronous, so verify membership
+and leaders before proceeding with maintenance.
+`/v1/task/compact` preserves its `compact ok` success body as `text/plain`;
+structured error bodies use `application/json`, including follower refusals.
+
 ### Metrics
 
 ```bash

hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/StoreNodeService.java

@@ -336,6 +336,11 @@ public synchronized Metapb.Store updateStore(Metapb.Store store) throws PDExcept
      * @throws PDException
      */
     public synchronized void storeTurnoff(Metapb.Store store) throws PDException {
+        storeTurnoffWithResult(store);
+    }
+
+    public synchronized List<Integer> storeTurnoffWithResult(Metapb.Store store) throws PDException {
+        List<Integer> changed = new ArrayList<>();
         // Traverse ShardGroup,redistribution
         for (Metapb.ShardGroup group : getShardGroupsByStore(store.getId())) {
             Metapb.ShardGroup.Builder builder = Metapb.ShardGroup.newBuilder(group);
@@ -345,8 +350,12 @@ public synchronized void storeTurnoff(Metapb.Store store) throws PDException {
                     builder.addShards(shard);
                 }
             });
-            reallocShards(builder.build());
+            List<Metapb.Shard> shards = reallocShards(builder.build());
+            if (!group.getShardsList().equals(shards)) {
+                changed.add(group.getId());
+            }
         }
+        return changed;
     }
 
     /**

hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java

@@ -41,6 +41,7 @@
 import org.apache.hugegraph.pd.meta.TaskInfoMeta;
 import org.apache.hugegraph.pd.raft.RaftEngine;
 
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -249,14 +250,31 @@ public List<Metapb.Store> patrolStores() throws PDException {
      * of replicas in the shard group
      */
     public List<Metapb.Partition> patrolPartitions() throws PDException {
+        PartitionPatrolResult result = patrolPartitionsWithResult();
+        return result == null ? null : result.getPartitions();
+    }
+
+    @Getter
+    public static class PartitionPatrolResult {
+
+        private final List<Integer> reallocatedShardGroups = new ArrayList<>();
+        private final List<Metapb.Partition> partitions = new ArrayList<>();
+        private final Map<Integer, Long> offlineShardRequests = new HashMap<>();
+    }
+
+    public PartitionPatrolResult patrolPartitionsWithResult() throws PDException {
         if (!isLeader()) {
             return null;
         }
+        PartitionPatrolResult result = new PartitionPatrolResult();
 
         // If the number of replicas is inconsistent, reallocate replicas
         for (Metapb.ShardGroup group : storeService.getShardGroups()) {
             if (group.getShardsCount() != pdConfig.getPartition().getShardCount()) {
-                storeService.reallocShards(group);
+                List<Metapb.Shard> shards = storeService.reallocShards(group);
+                if (!group.getShardsList().equals(shards)) {
+                    result.reallocatedShardGroups.add(group.getId());
+                }
                 kvService.put(BALANCE_SHARD_KEY, "DOING", 180 * 1000);
             }
         }
@@ -273,13 +291,20 @@ public List<Metapb.Partition> patrolPartitions() throws PDException {
                 }
                 partIds.add(partition.getId());
 
-                storeService.storeTurnoff(pair.getValue());
+                for (Integer changed : storeService.storeTurnoffWithResult(pair.getValue())) {
+                    if (!result.reallocatedShardGroups.contains(changed)) {
+                        result.reallocatedShardGroups.add(changed);
+                    }
+                }
                 partitionService.shardOffline(partition, pair.getValue().getId());
+                // shardOffline logs its own failures; report a request, not a completed repair.
+                result.offlineShardRequests.put(partition.getId(), pair.getValue().getId());
+                result.partitions.add(partition);
             }
 
         }
 
-        return null;
+        return result;
     }
 
     /**

hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/rest/TaskAPI.java

@@ -20,12 +20,14 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.hugegraph.pd.TaskScheduleService.PartitionPatrolResult;
 import org.apache.hugegraph.pd.common.KVPair;
 import org.apache.hugegraph.pd.common.PDException;
 import org.apache.hugegraph.pd.grpc.Metapb;
 import org.apache.hugegraph.pd.service.PDRestService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.ResponseBody;
@@ -57,8 +59,11 @@ public String patrolStores() {
     @ResponseBody
     public String patrolPartitions() {
         try {
-            List<Metapb.Partition> partitions = pdRestService.patrolPartitions();
-            return toJSON(partitions, "partitions");
+            PartitionPatrolResult result = pdRestService.patrolPartitionsWithResult();
+            String partitions = toJSON(result.getPartitions(), "partitions");
+            return partitions.substring(0, partitions.length() - 1) +
+                   ",\"reallocatedShardGroups\":" + toJSON(result.getReallocatedShardGroups()) +
+                   ",\"offlineShardRequests\":" + toJSON(result.getOfflineShardRequests()) + "}";
         } catch (PDException e) {
             e.printStackTrace();
             return toJSON(e);
@@ -67,13 +72,16 @@ public String patrolPartitions() {
 
     @GetMapping(value = "/balancePartitions", produces = MediaType.APPLICATION_JSON_VALUE)
     @ResponseBody
-    public Map<Integer, KVPair<Long, Long>> balancePartitions() {
+    public String balancePartitions() {
         try {
             Map<Integer, KVPair<Long, Long>> partitions = pdRestService.balancePartitions();
-            return partitions;
+            if (partitions == null) {
+                return toJSON(Map.of("status", 0, "outcome", "skipped"));
+            }
+            return toJSON(partitions);
         } catch (PDException e) {
             e.printStackTrace();
-            return null;
+            return toJSON(e);
         }
     }
 
@@ -89,12 +97,6 @@ public String splitPartitions() {
         }
     }
 
-    // TODO: make the task routes tell a real run from a no-op and a follower answer (a follower
-    // returns an empty success without doing anything), and return a body that names what was
-    // scheduled. Part 1 (#3233) replaced the bare 500 with a refusal body; this is part 2. The
-    // Helm chart (helm/hugegraph) documents the leader-first sequence and the 180 s spacing in its
-    // README until then.
-    // https://github.com/apache/hugegraph/issues/3231
     @GetMapping(value = "/balanceLeaders", produces = MediaType.APPLICATION_JSON_VALUE)
     @ResponseBody
     public String balanceLeaders() {
@@ -108,8 +110,12 @@ public String balanceLeaders() {
     }
 
     @GetMapping(value = "/compact")
-    public String dbCompaction() throws PDException {
-        pdRestService.dbCompaction();
-        return "compact ok";
+    public ResponseEntity<String> dbCompaction() {
+        try {
+            pdRestService.dbCompaction();
+            return ResponseEntity.ok().contentType(MediaType.TEXT_PLAIN).body("compact ok");
+        } catch (PDException e) {
+            return ResponseEntity.ok().contentType(MediaType.APPLICATION_JSON).body(toJSON(e));
+        }
     }
 }

hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/PDRestService.java

@@ -40,6 +40,7 @@
 import org.apache.hugegraph.pd.grpc.discovery.RegisterInfo;
 import org.apache.hugegraph.pd.model.RegistryRestRequest;
 import org.apache.hugegraph.pd.model.RegistryRestResponse;
+import org.apache.hugegraph.pd.raft.RaftEngine;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -137,11 +138,38 @@ public Map<Integer, Metapb.ShardGroup> getShardGroupCache() {
     }
 
     public List<Metapb.Store> patrolStores() throws PDException {
-        return monitorService.patrolStores();
+        return runLeaderTask(() -> monitorService.patrolStores());
     }
 
     public List<Metapb.Partition> patrolPartitions() throws PDException {
-        return monitorService.patrolPartitions();
+        return runLeaderTask(() -> monitorService.patrolPartitions());
+    }
+
+    public TaskScheduleService.PartitionPatrolResult patrolPartitionsWithResult() throws PDException {
+        return runLeaderTask(() -> monitorService.patrolPartitionsWithResult());
+    }
+
+    private <T> T runLeaderTask(LeaderTask<T> task) throws PDException {
+        checkTaskLeader();
+        T result = task.run();
+        // Do not turn a scheduler's follower no-op into success if leadership changed mid-call.
+        checkTaskLeader();
+        return result;
+    }
+
+    private void checkTaskLeader() throws PDException {
+        RaftEngine engine = RaftEngine.getInstance();
+        if (!engine.isLeader()) {
+            throw new PDException(Pdpb.ErrorType.NOT_LEADER_VALUE,
+                                  "Task requires the PD leader; leader Raft address: " +
+                                  engine.getLeader());
+        }
+    }
+
+    @FunctionalInterface
+    private interface LeaderTask<T> {
+
+        T run() throws PDException;
     }
 
     public Metapb.PartitionStats getPartitionStats(String graphName, int partitionId) throws
@@ -154,11 +182,11 @@ public List<Metapb.PartitionStats> getPartitionStatus(String graphName) throws P
     }
 
     public Map<Integer, KVPair<Long, Long>> balancePartitions() throws PDException {
-        return monitorService.balancePartitionShard();
+        return runLeaderTask(() -> monitorService.balancePartitionShard());
     }
 
     public List<Metapb.Partition> splitPartitions() throws PDException {
-        return monitorService.autoSplitPartition();
+        return runLeaderTask(() -> monitorService.autoSplitPartition());
     }
 
     public List<Metapb.Store> getStoreStats(boolean isActive) throws PDException {
@@ -262,11 +290,17 @@ public List<Metapb.LogRecord> getPartitionLog(Long start, Long end) throws PDExc
     }
 
     public Map<Integer, Long> balancePartitionLeader() throws PDException {
-        return monitorService.balancePartitionLeader(true);
+        return runLeaderTask(() -> monitorService.balancePartitionLeader(true));
     }
 
     public void dbCompaction() throws PDException {
-        monitorService.dbCompaction("");
+        runLeaderTask(() -> {
+            if (!monitorService.dbCompaction("")) {
+                throw new PDException(Pdpb.ErrorType.NOT_LEADER_VALUE,
+                                      "Compaction was refused by the scheduler; retry on the PD leader");
+            }
+            return null;
+        });
     }
 
     public List<Metapb.Shard> getShardList(int partitionId) throws PDException {