Merge branch 'master' into task/improve-condition-query-semantics

Author: imbajin

.github/workflows/server-ci.yml

@@ -106,7 +106,11 @@ jobs:
 
       - name: Run api test
         run: |
-          $TRAVIS_DIR/run-api-test.sh $BACKEND $REPORT_DIR
+          if [ "$BACKEND" = "rocksdb" ]; then
+            $TRAVIS_DIR/run-api-test.sh $BACKEND $REPORT_DIR true
+          else
+            $TRAVIS_DIR/run-api-test.sh $BACKEND $REPORT_DIR
+          fi
 
       # TODO: disable raft test in normal PR due to the always timeout problem
       - name: Run raft test
@@ -180,7 +184,7 @@ jobs:
 
       - name: Run RocksDB API test
         run: |
-          $TRAVIS_DIR/run-api-test.sh $BACKEND $REPORT_DIR
+          $TRAVIS_DIR/run-api-test.sh $BACKEND $REPORT_DIR true
 
       - name: Show server log on failure
         if: failure()

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java

@@ -102,12 +102,13 @@ public Map<String, Object> list(@Context GraphManager manager,
             limit = NO_LIMIT;
             List<Id> idList = ids.stream().map(IdGenerator::of)
                                  .collect(Collectors.toList());
-            iter = scheduler.tasks(idList);
+            iter = scheduler.tasks(idList, false);
         } else {
             if (status == null) {
-                iter = scheduler.tasks(null, limit, page);
+                iter = scheduler.tasks(null, limit, page, false);
             } else {
-                iter = scheduler.tasks(parseStatus(status), limit, page);
+                iter = scheduler.tasks(parseStatus(status), limit, page,
+                                       false);
             }
         }
 
@@ -136,12 +137,17 @@ public Map<String, Object> get(@Context GraphManager manager,
                                    @Parameter(description = "The graph name")
                                    @PathParam("graph") String graph,
                                    @Parameter(description = "The task id")
-                                   @PathParam("id") long id) {
+                                   @PathParam("id") long id,
+                                   @Parameter(description = "Whether to load task result")
+                                   @DefaultValue("true")
+                                   @QueryParam("with_result")
+                                   boolean withResult) {
         LOG.debug("Graph [{}] get task: {}", graph, id);
 
         TaskScheduler scheduler = graph(manager, graphSpace, graph)
                 .taskScheduler();
-        return scheduler.task(IdGenerator.of(id)).asMap();
+        return scheduler.task(IdGenerator.of(id), withResult)
+                        .asMap(true, withResult);
     }
 
     @DELETE

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java

@@ -1326,8 +1326,14 @@ public <V> void save(HugeTask<V> task) {
 
         @Override
         public <V> HugeTask<V> task(Id id) {
+            return this.task(id, true);
+        }
+
+        @Override
+        public <V> HugeTask<V> task(Id id, boolean withResult) {
             return verifyTaskPermission(HugePermission.READ,
-                                        this.taskScheduler.task(id));
+                                        this.taskScheduler.task(id,
+                                                                withResult));
         }
 
         @Override
@@ -1336,18 +1342,36 @@ public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
                                         this.taskScheduler.tasks(ids));
         }
 
+        @Override
+        public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
+                                               boolean withResult) {
+            return verifyTaskPermission(HugePermission.READ,
+                                        this.taskScheduler.tasks(ids,
+                                                                 withResult));
+        }
+
         @Override
         public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
                                                long limit, String page) {
             Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(status,
-                                                                   limit, page);
+                                                                   limit,
+                                                                   page);
+            return verifyTaskPermission(HugePermission.READ, tasks);
+        }
+
+        @Override
+        public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
+                                               long limit, String page,
+                                               boolean withResult) {
+            Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(
+                    status, limit, page, withResult);
             return verifyTaskPermission(HugePermission.READ, tasks);
         }
 
         @Override
         public <V> HugeTask<V> delete(Id id, boolean force) {
             verifyTaskPermission(HugePermission.DELETE,
-                                 this.taskScheduler.task(id));
+                                 this.taskScheduler.task(id, false));
             return this.taskScheduler.delete(id, force);
         }
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java

@@ -52,10 +52,10 @@ public class CachedSchemaTransactionV2 extends SchemaTransactionV2 {
 
     // MetaDriver doesn't expose unlisten, register the meta listener once.
     // Lifecycle: this JVM-global flag is intentionally never reset by
-    // unlistenChanges() (the underlying gRPC watch is process-wide). If that
-    // watch is silently dropped after a transport reconnect, recovery is not
-    // automatic; resetMetaListenerForReconnect() is only a manual hook to let
-    // the next schema operation install a fresh watch.
+    // unlistenChanges() (the underlying gRPC watch is process-wide). The driver
+    // watch self-heals across transport reconnects (PdMetaDriver via KvClient,
+    // EtcdMetaDriver via Watch.Listener re-subscribe), so the subscription stays
+    // live and the flag staying true is correct.
     private static final AtomicBoolean metaEventListenerRegistered =
             new AtomicBoolean(false);
 
@@ -251,27 +251,6 @@ static <T> void handleSchemaCacheClearEvent(T response) {
         }
     }
 
-    /**
-     * Manually reset the JVM-global meta listener flag after detecting that
-     * the MetaManager transport reconnected and dropped the underlying gRPC
-     * watch. This method is not wired to a MetaManager/MetaDriver reconnect
-     * callback today; callers must invoke it explicitly after detecting that
-     * condition. Without such a manual reset {@link #metaEventListenerRegistered}
-     * would stay {@code true} forever and this JVM would stop receiving
-     * cross-node schema cache clear events with no error or warning.
-     *
-     * <p>TODO: wire this into MetaManager once it exposes a transport
-     * reconnect callback (e.g. {@code listenReconnect} /
-     * {@code onTransportReconnect}). Until then it must be invoked
-     * explicitly by code that detects the reconnect.
-     */
-    public static void resetMetaListenerForReconnect() {
-        if (metaEventListenerRegistered.compareAndSet(true, false)) {
-            LOG.warn("Schema cache clear meta listener lost on reconnect - " +
-                     "will re-register on next schema operation.");
-        }
-    }
-
     public void clearCache(boolean notify) {
         // Same TOCTOU ordering as clearSchemaCache(String): clear nameCache
         // first, then the array attachment, then idCache last.

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java

@@ -25,6 +25,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import org.apache.commons.io.FileUtils;
@@ -33,7 +36,9 @@
 import org.apache.hugegraph.meta.lock.LockResult;
 import org.apache.hugegraph.type.define.CollectionType;
 import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
 import org.apache.hugegraph.util.collection.CollectionFactory;
+import org.slf4j.Logger;
 
 import com.google.common.base.Strings;
 
@@ -42,6 +47,7 @@
 import io.etcd.jetcd.ClientBuilder;
 import io.etcd.jetcd.KV;
 import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.Watch;
 import io.etcd.jetcd.kv.GetResponse;
 import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
 import io.etcd.jetcd.options.DeleteOption;
@@ -57,9 +63,24 @@
 
 public class EtcdMetaDriver implements MetaDriver {
 
+    private static final Logger LOG = Log.logger(EtcdMetaDriver.class);
+
     private final Client client;
     private final EtcdDistributedLock lock;
 
+    // Re-subscribes a dropped watch off the jetcd callback thread; single
+    // daemon thread, process-lifetime (no close()), so JVM shutdown reclaims it.
+    private final ScheduledExecutorService reWatchExecutor =
+            Executors.newSingleThreadScheduledExecutor(r -> {
+                Thread thread = new Thread(r, "etcd-meta-rewatch");
+                thread.setDaemon(true);
+                return thread;
+            });
+
+    // Backoff before re-subscribing a dropped watch. Package-private and
+    // mutable only so tests can set it to 0; never reassigned in production.
+    long reWatchDelayMs = 1000L;
+
     public EtcdMetaDriver(String trustFile, String clientCertFile,
                           String clientKeyFile, Object... endpoints) {
         ClientBuilder builder = this.etcdMetaDriverBuilder(endpoints);
@@ -76,6 +97,13 @@ public EtcdMetaDriver(Object... endpoints) {
         this.lock = EtcdDistributedLock.getInstance(this.client);
     }
 
+    // Package-private constructor for tests: inject a mock Client and skip lock
+    // setup (watch tests never touch the distributed lock).
+    EtcdMetaDriver(Client client) {
+        this.client = client;
+        this.lock = null;
+    }
+
     private static ByteSequence toByteSequence(String content) {
         return ByteSequence.from(content.getBytes());
     }
@@ -303,9 +331,8 @@ public void unlock(String key, LockResult lockResult) {
     @SuppressWarnings("unchecked")
     @Override
     public <T> void listen(String key, Consumer<T> consumer) {
-
-        this.client.getWatchClient().watch(toByteSequence(key),
-                                           (Consumer<WatchResponse>) consumer);
+        this.watchKey(toByteSequence(key), WatchOption.DEFAULT,
+                      (Consumer<WatchResponse>) consumer);
     }
 
     /**
@@ -314,9 +341,63 @@ public <T> void listen(String key, Consumer<T> consumer) {
     @SuppressWarnings("unchecked")
     @Override
     public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
-        ByteSequence sequence = toByteSequence(prefix);
         WatchOption option = WatchOption.newBuilder().isPrefix(true).build();
-        this.client.getWatchClient().watch(sequence, option, (Consumer<WatchResponse>) consumer);
+        this.watchKey(toByteSequence(prefix), option,
+                      (Consumer<WatchResponse>) consumer);
+    }
+
+    /**
+     * Subscribe a watch that survives the terminal close jetcd cannot recover
+     * from. The bare {@code Consumer} overload discards {@code onError} and
+     * {@code onCompleted}, so a non-retryable failure silently drops the
+     * listener (issue #3036).
+     * <p>
+     * jetcd 0.5.9 ({@code WatchImpl.WatcherImpl.handleError}) already retries
+     * <em>retryable</em> errors itself: it notifies {@code onError} and then
+     * reschedules {@code resume()} on the same watcher. Re-subscribing from
+     * {@code onError} would therefore open a duplicate watch on every transient
+     * reconnect, so {@code onError} only logs here. A non-retryable error (or an
+     * explicit cancel) ends in {@code close()}, which removes the watcher and
+     * invokes {@code onCompleted}; that is the only point where the watch is
+     * truly gone, so re-subscribe happens there. The old watcher is already
+     * closed and removed, so the replacement is not a duplicate.
+     */
+    private void watchKey(ByteSequence key, WatchOption option,
+                          Consumer<WatchResponse> consumer) {
+        Watch.Listener listener = Watch.listener(
+                consumer,
+                throwable -> LOG.warn("etcd meta watch error for key '{}', " +
+                                      "jetcd will retry if recoverable",
+                                      key.toString(Charset.defaultCharset()),
+                                      throwable),
+                () -> this.scheduleReWatch(key, option, consumer));
+        this.client.getWatchClient().watch(key, option, listener);
+    }
 
+    private void scheduleReWatch(ByteSequence key, WatchOption option,
+                                 Consumer<WatchResponse> consumer) {
+        this.reWatchExecutor.schedule(() -> this.reWatch(key, option, consumer),
+                                      this.reWatchDelayMs, TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * Re-establish a watch dropped by a terminal close. If the re-subscribe
+     * itself fails (e.g. the endpoint is still unreachable), it is retried with
+     * the same backoff instead of giving up, otherwise a single failed attempt
+     * would lose the listener permanently.
+     */
+    private void reWatch(ByteSequence key, WatchOption option,
+                         Consumer<WatchResponse> consumer) {
+        try {
+            LOG.info("Re-establishing etcd meta watch for key '{}'",
+                     key.toString(Charset.defaultCharset()));
+            this.watchKey(key, option, consumer);
+        } catch (Exception e) {
+            LOG.warn("Failed to re-establish etcd meta watch for key '{}', " +
+                     "retrying in {} ms",
+                     key.toString(Charset.defaultCharset()),
+                     this.reWatchDelayMs, e);
+            this.scheduleReWatch(key, option, consumer);
+        }
     }
 }

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java

@@ -306,10 +306,14 @@ protected <V> HugeTask<V> deleteFromDB(Id id) {
             Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
             HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
             if (vertex == null) {
+                this.deleteTaskResultFromTx(id);
                 return null;
             }
-            HugeTask<V> result = HugeTask.fromVertex(vertex);
-            this.tx().removeVertex(vertex);
+            HugeTask<V> result = HugeTask.fromVertex(vertex, false);
+            // Keep the task vertex as a retryable tombstone until its result
+            // vertex is removed; cronSchedule() can rediscover DELETING tasks.
+            this.deleteTaskResultFromTx(id);
+            this.tx().removeTaskVertex(vertex);
             return result;
         });
     }
@@ -322,6 +326,12 @@ public <V> HugeTask<V> delete(Id id, boolean force) {
             this.updateStatus(id, null, TaskStatus.DELETING);
             return null;
         } else {
+            HugeTask<?> task = this.taskWithoutResult(id);
+            if (task != null && task.status() != TaskStatus.DELETING) {
+                initTaskParams(task);
+                task.overwriteStatus(TaskStatus.DELETING);
+                this.save(task);
+            }
             return this.deleteFromDB(id);
         }
     }
@@ -587,7 +597,7 @@ private void unlockTask(String taskId, LockResult lockResult) {
         }
     }
 
-    private boolean isLockedTask(String taskId) {
+    protected boolean isLockedTask(String taskId) {
         return MetaManager.instance().isLockedTask(graphSpace,
                                                    graphName, taskId);
     }
@@ -629,7 +639,7 @@ public void run() {
                     // 1. start task can be from schedule() & cronSchedule()
                     // 2. recheck the status of task, in case one same task
                     // called by both methods at same time;
-                    HugeTask<Object> queryTask = task(this.task.id());
+                    HugeTask<Object> queryTask = task(this.task.id(), false);
                     if (queryTask != null &&
                         !TaskStatus.NEW.equals(queryTask.status())) {
                         return;