refactor(server): close retained graphs on shutdown

docs/server-runtime.md

@@ -498,6 +498,17 @@ drain does not queue transaction cleanup behind that scheduler's blocked worker.
 distributed task dispatch is distinct from completing drain: a timeout does not make
 the next close return success.
 
+During Server shutdown, the task manager waits for task workers to finish, then retries
+closing graphs retained in its existing scheduler registry before stopping the task
+database worker. This gives pending tasks time to persist completion and releases the
+task database owner's transactions on that owner's thread. Each retained graph gets one
+close attempt per shutdown call. A close failure remains visible and keeps the database
+executor available for a later explicit retry; it does not start background recovery.
+A terminated cron executor already satisfies the scheduler's dispatch barrier. Every
+executor wait, including an explicit retry after a timeout, checks actual termination;
+shutdown requested alone is insufficient. Store-close notifications finish before the
+shared event executor is stopped.
+
 Once these logical checks pass, graph close releases shared cache listeners, invokes
 `provider.close()` and destroys graph locks even if an earlier cleanup step reported an error. The error
 remains visible to the caller. Logical graph closure does not prove that a failed
@@ -511,7 +522,8 @@ Regression sources:
 - [`StandardTaskSchedulerTxTest`](../hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/task/StandardTaskSchedulerTxTest.java)
   exercises task worker transaction cleanup.
 - [`HugeFactoryTest`](../hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java)
-  exercises graph close failure continuation and distributed drain retry.
+  exercises graph close failure continuation, distributed drain retry and Server stop
+  with a task that outlives the early graph-close attempts.
 
 ## Compatibility evidence and deployment checks
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeFactory.java

@@ -187,10 +187,10 @@ public static void shutdown(long timeout, boolean ignoreException) {
         }
         try {
 
+            TaskManager.instance().shutdown(timeout);
             if (!EventHub.destroy(timeout)) {
                 throw new TimeoutException(timeout + "s");
             }
-            TaskManager.instance().shutdown(timeout);
             OltpTraverser.destroy();
         } catch (Throwable e) {
             LOG.error("Error while shutdown", e);

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

@@ -622,25 +622,28 @@ public boolean close() {
             cronFuture.cancel(false);
         }
 
-        // Wait behind the scheduler thread to ensure any running cron task is completed
-        try {
-            Future<?> barrier = this.schedulerExecutor.submit(() -> {
-                // pass
-            });
-            barrier.get(schedulePeriod + 5, TimeUnit.SECONDS);
-        } catch (TimeoutException e) {
-            LOG.warn("Cron task did not complete in time when closing scheduler");
-            return false;
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            LOG.warn("Interrupted while waiting for cron task to complete", e);
-            return false;
-        } catch (ExecutionException e) {
-            LOG.warn("Exception while waiting for cron task to complete", e);
-            return false;
-        } catch (RejectedExecutionException e) {
-            LOG.warn("Scheduler executor was stopped before its graphs drained", e);
-            return false;
+        // A terminated executor has already completed all its cron work.
+        // Otherwise wait behind that work before closing its owners.
+        if (!this.schedulerExecutor.isTerminated()) {
+            try {
+                Future<?> barrier = this.schedulerExecutor.submit(() -> {
+                    // pass
+                });
+                barrier.get(schedulePeriod + 5, TimeUnit.SECONDS);
+            } catch (TimeoutException e) {
+                LOG.warn("Cron task did not complete in time when closing scheduler");
+                return false;
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                LOG.warn("Interrupted while waiting for cron task to complete", e);
+                return false;
+            } catch (ExecutionException e) {
+                LOG.warn("Exception while waiting for cron task to complete", e);
+                return false;
+            } catch (RejectedExecutionException e) {
+                LOG.warn("Scheduler executor was stopped before its graphs drained", e);
+                return false;
+            }
         }
 
         // cancel all running tasks

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

@@ -17,6 +17,7 @@
 
 package org.apache.hugegraph.task;
 
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.Callable;
@@ -255,69 +256,50 @@ public void shutdown(long timeout) {
             LOG.warn("Stopping task executors with {} schedulers still registered", this.schedulers.size());
         }
 
-        Throwable ex = null;
-        boolean terminated = this.distributedSchedulerExecutor.isTerminated();
-        final TimeUnit unit = TimeUnit.SECONDS;
-
-        if (!this.distributedSchedulerExecutor.isShutdown()) {
-            this.distributedSchedulerExecutor.shutdown();
-            try {
-                terminated = this.distributedSchedulerExecutor.awaitTermination(timeout, unit);
-            } catch (Throwable e) {
-                ex = e;
-            }
-        }
-
-        if (terminated && !this.taskExecutor.isShutdown()) {
-            this.taskExecutor.shutdown();
-            try {
-                terminated = this.taskExecutor.awaitTermination(timeout, unit);
-            } catch (Throwable e) {
-                ex = e;
-            }
-        }
-
-        if (terminated && !this.taskDbExecutor.isShutdown()) {
-            this.taskDbExecutor.shutdown();
-            try {
-                terminated = this.taskDbExecutor.awaitTermination(timeout, unit);
-            } catch (Throwable e) {
-                ex = e;
-            }
+        try {
+            shutdownExecutor(this.distributedSchedulerExecutor, timeout);
+            shutdownExecutor(this.taskExecutor, timeout);
+            // Early REST/Gremlin closes can retain graphs while tasks finish.
+            // Their DB owners must close before the owner executor stops.
+            this.closeRetainedGraphs();
+            shutdownExecutor(this.taskDbExecutor, timeout);
+            shutdownExecutor(this.ephemeralTaskExecutor, timeout);
+            shutdownExecutor(this.schemaTaskExecutor, timeout);
+            shutdownExecutor(this.olapTaskExecutor, timeout);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new HugeException("Failed to wait for TaskScheduler", e);
+        } catch (TimeoutException e) {
+            throw new HugeException("Failed to wait for TaskScheduler", e);
         }
+    }
 
-        if (terminated && !this.ephemeralTaskExecutor.isShutdown()) {
-            this.ephemeralTaskExecutor.shutdown();
-            try {
-                terminated = this.ephemeralTaskExecutor.awaitTermination(timeout, unit);
-            } catch (Throwable e) {
-                ex = e;
-            }
+    private static void shutdownExecutor(ExecutorService executor, long timeout)
+                                         throws InterruptedException, TimeoutException {
+        if (!executor.isShutdown()) {
+            executor.shutdown();
         }
-
-        if (terminated && !this.schemaTaskExecutor.isShutdown()) {
-            this.schemaTaskExecutor.shutdown();
-            try {
-                terminated = this.schemaTaskExecutor.awaitTermination(timeout, unit);
-            } catch (Throwable e) {
-                ex = e;
-            }
+        if (!executor.isTerminated() && !executor.awaitTermination(timeout, TimeUnit.SECONDS)) {
+            throw new TimeoutException(timeout + "s");
         }
+    }
 
-        if (terminated && !this.olapTaskExecutor.isShutdown()) {
-            this.olapTaskExecutor.shutdown();
+    private void closeRetainedGraphs() {
+        Throwable failure = null;
+        for (HugeGraphParams graph : new ArrayList<>(this.schedulers.keySet())) {
             try {
-                terminated = this.olapTaskExecutor.awaitTermination(timeout, unit);
-            } catch (Throwable e) {
-                ex = e;
+                graph.graph().close();
+            } catch (Exception | Error error) {
+                if (failure == null) {
+                    failure = error;
+                } else if (failure != error) {
+                    failure.addSuppressed(error);
+                }
             }
         }
-
-        if (!terminated) {
-            ex = new TimeoutException(timeout + "s");
-        }
-        if (ex != null) {
-            throw new HugeException("Failed to wait for TaskScheduler", ex);
+        if (failure != null) {
+            // Keep the registry and DB executor available for an explicit retry.
+            throw new HugeException("Failed to close retained graphs before task DB shutdown", failure);
         }
     }