fix: release query resources and drain Store RPCs

.github/workflows/pd-store-ci.yml

@@ -8,6 +8,21 @@ permissions:
   contents: read
 
 jobs:
+  store-stop-shell-test:
+    permissions:
+      contents: read
+    runs-on: ubuntu-24.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+
+      - name: Verify Store stop failure propagation and PID retention
+        run: |
+          docker run --rm -v "$GITHUB_WORKSPACE:/work:ro" -w /work ubuntu:24.04 \
+            bash hugegraph-server/hugegraph-dist/src/assembly/travis/test-stop-hugegraph-store.sh
+
   rocksdb-compatibility:
     permissions:
       contents: read
@@ -431,6 +446,11 @@ jobs:
           mvn test -pl hugegraph-store/hg-store-test -am \
             -P store-core-test -Djacoco.sessionId=store-core-test
 
+      - name: Run server shutdown test
+        run: |
+          mvn test -pl hugegraph-store/hg-store-test -am \
+            -P store-server-test -Djacoco.sessionId=store-server-test
+
       - name: Generate aggregate coverage report
         run: |
           mvn verify -pl hugegraph-store/hg-store-test -am -P jacoco \
@@ -450,6 +470,8 @@ jobs:
             --require-test-report \
             "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.core.CoreSuiteTest.xml" \
             --require-test-report \
+            "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.service.ServerSuiteTest.xml" \
+            --require-test-report \
             "hugegraph-store/hg-store-node/target/surefire-reports/TEST-org.apache.hugegraph.store.business.StoredRowIngressTest.xml" \
             --require-covered-group hg-store-common \
             --require-covered-group hg-store-client \
@@ -460,6 +482,7 @@ jobs:
             --require-session store-rocksdb-test \
             --require-session store-raftcore-test \
             --require-session store-core-test \
+            --require-session store-server-test \
             --require-session store-node-test \
             "$REPORT_FILE" \
             hg-store-grpc hg-store-common hg-store-client hg-store-rocksdb hg-store-core

README.md

@@ -316,9 +316,6 @@ gremlin> :> g.V().limit(5)
 ```
 
 For comprehensive documentation, visit the [HugeGraph Documentation](https://hugegraph.apache.org/docs/).
-
-See [standalone RocksDB snapshot recovery](docs/rocksdb-recovery.md) before restoring data or mounting store directories.
-
 For an existing deployment, read the [RocksDB upgrade guidance](docs/rocksdb-upgrade.md) before upgrading the storage runtime.
 
 </details>

docs/transaction-lifecycle.md

@@ -0,0 +1,30 @@
+# Request transactions and Store shutdown
+
+Server releases the current thread's graph transactions when a REST request
+finishes and when an authenticated context task exits. Cleanup visits every
+registered graph even when one graph fails to close, and reports failures.
+Cleanup explicitly rolls back unfinished writes, including when `onClose(COMMIT)`
+was configured, then releases backend transactions and resets thread-local
+transaction behavior and listeners. Applications must commit successful writes
+before leaving the request/task boundary. Closing an OLTP traverser preserves
+the caller's transaction so that the caller can still commit or roll it back.
+
+The cleanup also applies when schema caches have become cold. Deleting an auth
+project or user must remove its associated access or belong edges while
+preserving unrelated relationships. The special OLAP vertex path remains
+separate; negative schema IDs alone do not identify edges that can be skipped.
+
+Store shutdown refuses new RPCs, cancels active RPCs and scans, and waits for their callbacks and scan/TTL workers before Spring destroys the Store engine and its databases. A failed aggregate-query response callback is logged without skipping cancellation or cleanup waits for other queries. Cancellation may fail in-flight requests; stop writes and check their outcomes before planned maintenance. This is not a guarantee that every in-flight request drains successfully or that a leader transfers without a failover interval.
+
+If a callback cannot finish, or an ordinary scan or bidirectional aggregate query fails to release its plan or partition iterators, shutdown remains pending rather than closing its database underneath it. This also includes iterators discarded while advancing past empty partitions, initializing a sequential scan, or counting rows. RocksDB iterators retain the first failure during automatic or explicit close so that later cleanup cannot hide it; concurrent close calls wait for the release attempt to finish. All scan entry points, including one-shot scans, retain failed cleanup independently of RPC termination and executor shutdown. Failed iterator releases are attempted once and remain diagnostic shutdown blockers. Bidirectional aggregate queries report cleanup failures to the client and log that shutdown is blocked; their final success batch is sent only after cleanup succeeds. The distribution stop script waits up to 30 seconds, returns a nonzero status on timeout, and retains the PID file for diagnosis. Inspect logs and thread dumps before retrying. Do not add a concurrent shutdown hook that closes the same databases.
+
+If shutdown cancellation wins while a one-shot scan releases its iterator, the
+response terminates with `CANCELLED`; it does not report successful completion
+without its result. Iterator cleanup still finishes before the scan unregisters.
+
+A normal aggregate-query request half-close ends feedback without cancelling already permitted work. A subsequent transport cancellation or deadline still interrupts the workers and releases their resources. A batch-scan RPC accepts one initial query; repeated query requests are ignored before allocating another iterator, including after its final batch. If the remaining feedback credit cannot finish the query, the server returns an explicit query error. Scan task rejection reports `UNAVAILABLE` during shutdown and `RESOURCE_EXHAUSTED` when the running scan pool is full.
+
+See the [Store shutdown instructions](../hugegraph-store/README.md#stopping-a-store-node)
+and the Server [module test guidance](../hugegraph-server/AGENTS.md#tests).
+
+Shared schema and element caches retain their invalidation listeners until the graph closes. Request cleanup releases backend leases while preserving those graph caches and their schema identity.

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

@@ -185,7 +185,7 @@ private static void registerPrivateActions() {
                                           "this$0");
         Reflection.registerFieldsToFilter(HugeGraphAuthProxy.Context.class, "ADMIN", "user");
         Reflection.registerFieldsToFilter(HugeGraphAuthProxy.ContextTask.class, "runner",
-                                          "context");
+                                          "cleanup", "context");
         Reflection.registerFieldsToFilter(StandardHugeGraph.class, "LOG", "started", "closed",
                                           "mode", "variables", "name", "params", "configuration",
                                           "schemaEventHub", "graphEventHub", "indexEventHub",
@@ -203,7 +203,8 @@ private static void registerPrivateActions() {
                                            "access$14", "access$15", "access$16", "access$17",
                                            "access$18", "serializer", "loadSchemaStore",
                                            "loadSystemStore", "loadGraphStore", "closeTx",
-                                           "analyzer", "serverInfoManager", "reloadRamtable",
+                                           "closeCurrentThreadTransaction", "analyzer",
+                                           "serverInfoManager", "reloadRamtable",
                                            "reloadRamtable", "access$19", "access$20", "access$21");
         Reflection.registerFieldsToFilter(
                 loadClass("org.apache.hugegraph.StandardHugeGraph$StandardHugeGraphParams"),
@@ -298,8 +299,10 @@ private static void registerPrivateActions() {
                                            "autoCommit", "beforeRead", "afterWrite", "afterRead",
                                            "commitMutation2Backend", "checkOwnerThread", "doAction",
                                            "store", "reset");
-        Reflection.registerFieldsToFilter(HugeFactory.class, "LOG", "NAME_REGEX", "graphs");
-        Reflection.registerMethodsToFilter(HugeFactory.class, "lambda$0");
+        Reflection.registerFieldsToFilter(HugeFactory.class, "LOG", "NAME_REGEX", "graphs",
+                                          "GRAPHS");
+        Reflection.registerMethodsToFilter(HugeFactory.class, "lambda$0",
+                                           "closeCurrentThreadTransactions");
         Reflection.registerFieldsToFilter(SchemaElement.class, "graph", "id", "name", "userdata",
                                           "status");
         Reflection.registerFieldsToFilter(HugeVertex.class, "EMPTY_SET", "id", "label", "edges",

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

@@ -1291,11 +1291,17 @@ public User user() {
     static class ContextTask implements Runnable {
 
         private final Runnable runner;
+        private final Runnable cleanup;
         private final Context context;
 
         public ContextTask(Runnable runner) {
+            this(runner, HugeFactory::closeCurrentThreadTransactions);
+        }
+
+        ContextTask(Runnable runner, Runnable cleanup) {
             this.context = getContext();
             this.runner = runner;
+            this.cleanup = cleanup;
         }
 
         @Override
@@ -1305,7 +1311,7 @@ public void run() {
                 this.runner.run();
             } finally {
                 try {
-                    HugeFactory.closeCurrentThreadTransactions();
+                    this.cleanup.run();
                 } catch (Throwable e) {
                     LOG.error("Failed to close Gremlin worker transactions", e);
                 } finally {

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

@@ -22,6 +22,7 @@
 import java.io.File;
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -118,6 +119,11 @@ public static void closeCurrentThreadTransactions() {
         synchronized (HugeFactory.class) {
             graphs = new ArrayList<>(GRAPHS.values());
         }
+        closeCurrentThreadTransactions(graphs);
+    }
+
+    static void closeCurrentThreadTransactions(
+                Collection<? extends StandardHugeGraph> graphs) {
         Throwable failure = null;
         for (StandardHugeGraph graph : graphs) {
             try {

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

@@ -1361,22 +1361,25 @@ public void rollback() {
         }
 
         public void close() {
-            try {
-                this.graphTx.close();
-            } catch (Exception e) {
-                LOG.error("Failed to close GraphTransaction", e);
+            Throwable failure = null;
+            for (Runnable close : new Runnable[]{this.graphTx::close,
+                                                   this.systemTx::close,
+                                                   this.schemaTx::close}) {
+                try {
+                    close.run();
+                } catch (RuntimeException | Error error) {
+                    if (failure == null) {
+                        failure = error;
+                    } else if (failure != error) {
+                        failure.addSuppressed(error);
+                    }
+                }
             }
-
-            try {
-                this.systemTx.close();
-            } catch (Exception e) {
-                LOG.error("Failed to close SystemTransaction", e);
+            if (failure instanceof Error) {
+                throw (Error) failure;
             }
-
-            try {
-                this.schemaTx.close();
-            } catch (Exception e) {
-                LOG.error("Failed to close SchemaTransaction", e);
+            if (failure != null) {
+                throw (RuntimeException) failure;
             }
         }
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendSessionPool.java

@@ -56,14 +56,17 @@ public HugeConfig config() {
     public final BackendSession getOrNewSession() {
         BackendSession session = this.threadLocalSession.get();
         if (session == null) {
-            session = this.newSession();
-            assert session != null;
-            this.threadLocalSession.set(session);
-            assert !this.sessions.containsKey(Thread.currentThread().getId());
-            this.sessions.put(Thread.currentThread().getId(), session);
-            int sessionCount = this.sessionCount.incrementAndGet();
-            LOG.debug("Now(after connect({})) session count is: {}",
-                      this, sessionCount);
+            // Serialize new borrowers with the last-session native close.
+            synchronized (this) {
+                session = this.newSession();
+                assert session != null;
+                this.threadLocalSession.set(session);
+                assert !this.sessions.containsKey(Thread.currentThread().getId());
+                this.sessions.put(Thread.currentThread().getId(), session);
+                int sessionCount = this.sessionCount.incrementAndGet();
+                LOG.debug("Now(after connect({})) session count is: {}",
+                          this, sessionCount);
+            }
         } else {
             this.detectSession(session);
         }
@@ -131,7 +134,7 @@ public void forceResetSessions() {
         }
     }
 
-    public boolean close() {
+    public synchronized boolean close() {
         Pair<Integer, Integer> result = Pair.of(-1, -1);
         try {
             result = this.closeSession();

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java

@@ -765,7 +765,7 @@ public Iterator<Vertex> queryTaskInfos(Query query) {
     }
 
     public Iterator<Vertex> queryTaskInfos(Object... vertexIds) {
-        if (this.graph().backendStoreFeatures().supportsTaskAndServerVertex()) {
+        if (this.storeFeatures().supportsTaskAndServerVertex()) {
             return this.queryVerticesByIds(vertexIds, false, false,
                                            HugeType.TASK);
         }

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

@@ -596,7 +596,7 @@ private <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
                                                 boolean withResult) {
         return this.call(() -> {
             ConditionQuery query;
-            if (this.graph.backendStoreFeatures().supportsTaskAndServerVertex()) {
+            if (this.tx().storeFeatures().supportsTaskAndServerVertex()) {
                 query = new ConditionQuery(HugeType.TASK);
             } else {
                 query = new ConditionQuery(HugeType.VERTEX);

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java

@@ -73,7 +73,7 @@ protected OltpTraverser(HugeGraph graph) {
 
     @Override
     public void close() {
-        // pass
+        // The graph's thread-local transaction belongs to the caller.
     }
 
     public static void destroy() {

hugegraph-server/hugegraph-dist/src/assembly/travis/run-native-runtime-smoke-test.sh

@@ -39,7 +39,7 @@ cleanup() {
     local status=$?
     trap - EXIT
     if [[ "$SERVER_START_ATTEMPTED" == "true" ]]; then
-        "$SERVER_DIR/bin/stop-hugegraph.sh" -m false >/dev/null 2>&1 || status=1
+        "$SERVER_DIR/bin/stop-hugegraph.sh" -m false || status=1
     fi
     rm -rf "$WORK_DIR"
     exit "$status"

hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh

@@ -430,8 +430,8 @@ def validation_command(job):
 
 def reports_for_option(job, option):
     pattern = re.escape(option) + (
-        r'\s+\\?\s*"\$TEST_REPORT_DIR/'
-        r'(TEST-[A-Za-z0-9_.]+SuiteTest[.]xml)"'
+        r'\s+\\?\s*"[^"]*/'
+        r'(TEST-[A-Za-z0-9_.]+Test[.]xml)"'
     )
     return set(re.findall(pattern, validation_command(job)))
 
@@ -492,9 +492,12 @@ assert_order(store_job, [
     "-P store-client-test -Djacoco.sessionId=store-client-test",
     "-P store-rocksdb-test -Djacoco.sessionId=store-rocksdb-test",
     "-P store-raftcore-test -Djacoco.sessionId=store-raftcore-test",
+    "-P store-core-test -Djacoco.sessionId=store-core-test",
+    "-P store-server-test -Djacoco.sessionId=store-server-test",
     "mvn verify", "--require-session store-common-test",
     "--require-session store-client-test", "--require-session store-rocksdb-test",
-    "--require-session store-raftcore-test", "codecov/codecov-action",
+    "--require-session store-raftcore-test", "--require-session store-core-test",
+    "--require-session store-server-test", "codecov/codecov-action",
 ])
 assert store_job.count("mvn clean") == 1
 assert "hugegraph-store/hg-store-test/target/site/jacoco/jacoco.xml" in store_job
@@ -504,16 +507,19 @@ assert "mvn verify -pl hugegraph-store/hg-store-test -am -P jacoco \\ " \
        "-DskipTests -Deditorconfig.skip=true -ntp" in " ".join(store_job.split())
 assert selected_profiles(store_job, "store") == {
     "store-common-test", "store-client-test", "store-rocksdb-test",
-    "store-raftcore-test", "store-core-test",
+    "store-raftcore-test", "store-core-test", "store-server-test",
 }
 assert reports_for_option(store_job, "--require-test-report") == {
     "TEST-org.apache.hugegraph.store.common.CommonSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.client.ClientSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.rocksdb.RocksDbSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.raftcore.RaftSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.core.CoreSuiteTest.xml",
+    "TEST-org.apache.hugegraph.store.service.ServerSuiteTest.xml",
+    "TEST-org.apache.hugegraph.store.business.StoredRowIngressTest.xml",
 }
 assert not reports_for_option(store_job, "--require-suite-report")
+assert values_for_option(store_job, "--require-session") == (selected_profiles(store_job, "store") | {"store-node-test"})
 assert values_for_option(store_job, "--require-covered-group") == {
     "hg-store-common", "hg-store-client", "hg-store-rocksdb", "hg-store-core",
 }

hugegraph-server/hugegraph-dist/src/assembly/travis/test-stop-hugegraph-store.sh

@@ -0,0 +1,42 @@
+#!/bin/bash
+#
+# 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.
+#
+# Verify stop failure propagation and PID retention without signalling real processes.
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../../.." && pwd)"
+SCRIPT="$ROOT/hugegraph-store/hg-store-dist/src/assembly/static/bin/stop-hugegraph-store.sh"
+FIXTURE=$(mktemp -d)
+trap 'rm -rf "$FIXTURE"' EXIT
+mkdir -p "$FIXTURE/bin"
+cp "$SCRIPT" "$FIXTURE/bin/stop-hugegraph-store.sh"
+cat > "$FIXTURE/bin/util.sh" <<'UTIL'
+kill_process_and_wait() {
+    [[ "$1" == HugeGraphStoreServer && "$2" == 12345 && "$3" == 30 ]] || return 99
+    return "$STOP_RESULT"
+}
+UTIL
+printf '12345\n' > "$FIXTURE/bin/pid"
+if STOP_RESULT=1 bash "$FIXTURE/bin/stop-hugegraph-store.sh"; then
+    echo "Stop timeout incorrectly returned success" >&2
+    exit 1
+fi
+[[ "$(cat "$FIXTURE/bin/pid")" == 12345 ]]
+STOP_RESULT=0 bash "$FIXTURE/bin/stop-hugegraph-store.sh"
+[[ ! -e "$FIXTURE/bin/pid" ]]
+STOP_RESULT=0 bash "$FIXTURE/bin/stop-hugegraph-store.sh"
+echo "store-stop-contract-ok"