refactor(store): drain TTL native resources on shutdown

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

@@ -8,6 +8,19 @@ 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: bash hugegraph-server/hugegraph-dist/src/assembly/travis/test-stop-hugegraph-store.sh
+
   rocksdb-compatibility:
     permissions:
       contents: read
@@ -431,7 +444,7 @@ jobs:
           mvn test -pl hugegraph-store/hg-store-test -am \
             -P store-core-test -Djacoco.sessionId=store-core-test
 
-      - name: Run server lifecycle test
+      - name: Run server lifecycle and shutdown tests
         run: |
           mvn test -pl hugegraph-store/hg-store-test -am \
             -P store-server-test -Djacoco.sessionId=store-server-test

docs/storage-lifecycle.md

@@ -12,8 +12,8 @@ atomic restore across all databases in a graph.
 
 - [RPC and Scan ownership](#rpc-and-scan-ownership): final batches, half-close,
   cancellation, iterator cleanup and callback dispatch.
-- [Shutdown coordination](#shutdown-coordination): admission, worker drain and
-  application callbacks before database teardown.
+- [Shutdown coordination](#shutdown-coordination): admission, worker drain, TTL cleanup
+  and [Store node shutdown](#store-node-shutdown).
 - [Standalone snapshot recovery](#standalone-snapshot-recovery): pending records,
   checkpoint identity, WAL paths and safe retry.
 - [Storage provider selection and ToplingDB](#storage-provider-selection-and-toplingdb):
@@ -114,6 +114,38 @@ releases graph database and schema resources. Keep this ordering when modifying
 database teardown. Request drain and database release have different responsibilities;
 reuse their existing owners instead of adding a parallel registry.
 
+### Store node shutdown
+
+Spring coordinates Store shutdown: new RPCs are refused, active requests are cancelled,
+and their application callbacks and scan workers are drained before database teardown.
+The shutdown coordinator continues cancellation and independent cleanup even if one
+callback fails. It reuses the existing per-call owners.
+
+TTL cleanup closes the scan before returning its borrowed RocksDB session lease. Both
+releases are attempted when either fails. The first cleanup failure is retained, later
+failures are attached, and completion is signalled after these release attempts.
+Executor termination alone does not establish that native cleanup succeeded.
+
+Shutdown waits for TTL cleanup as well as ordinary request callbacks. Failed native
+releases remain diagnostic drain blockers. The engine stops heartbeat producers and
+joins partition Raft services before releasing native databases. Do not add a concurrent
+JVM shutdown hook that closes those databases independently of this drain.
+
+Cancellation can fail in-flight writes. Before planned maintenance, stop new writes and
+check outstanding write outcomes.
+
+For an unpacked distribution, invoke its normal stop script:
+
+```bash
+bash bin/stop-hugegraph-store.sh
+```
+
+The script waits up to 30 seconds for process exit. A timeout returns a nonzero status
+and retains `bin/pid` for diagnosis; it does not force-kill the process or remove data.
+A stuck callback or failed iterator close can keep shutdown pending. Inspect Store logs
+and a JVM thread dump to identify the remaining owner before retrying. Preserve the data,
+WAL and Raft directories, and do not bypass the resource drain to make the process exit.
+
 ## Standalone snapshot recovery
 
 ### Installation and pending metadata

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-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"

hugegraph-store/README.md

@@ -397,10 +397,15 @@ For Docker and Kubernetes deployment details, see [Deployment Guide](docs/deploy
 
 ---
 
+## Stopping a Store node
+
+See [Store-wide shutdown](../docs/storage-lifecycle.md#store-node-shutdown) for stop commands,
+timeout handling, diagnostics and resource drain behavior.
+
 ## Documentation
 
-See [RPC and Scan lifecycle](../docs/storage-lifecycle.md#rpc-and-scan-ownership) for request ownership,
-cancellation and feedback semantics.
+See the [storage lifecycle guide](../docs/storage-lifecycle.md) for request ownership,
+shutdown, recovery and provider configuration.
 
 Comprehensive documentation for HugeGraph Store:
 

hugegraph-store/hg-store-dist/src/assembly/static/bin/stop-hugegraph-store.sh

@@ -41,8 +41,8 @@ if [ ! -f ${PID_FILE} ]; then
 fi
 
 PID=`cat $PID_FILE`
-kill_process_and_wait "HugeGraphStoreServer" "$PID" "$SERVER_SHUTDOWN_TIMEOUT_S"
-
-if [ $? -eq 0 ]; then
+if kill_process_and_wait "HugeGraphStoreServer" "$PID" "$SERVER_SHUTDOWN_TIMEOUT_S"; then
     rm "$PID_FILE"
+else
+    exit 1
 fi

hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/listener/ContextClosedListener.java

@@ -69,13 +69,9 @@ public void onApplicationEvent(ContextClosedEvent event) {
         }
         this.grpcServers.forEach(Server::shutdownNow);
         if (cleaner != null) {
-            // TTL failure ownership and worker drain are added by the native shutdown PR.
-            if (cleaner.getExecutor() != null) {
-                cleaner.getExecutor().shutdownNow();
-            }
-            if (cleaner.getScheduler() != null) {
-                cleaner.getScheduler().shutdownNow();
-            }
+            // The scheduler can create the worker pool while a job is starting.
+            stopAndWait(cleaner.getScheduler(), "TTL scheduler");
+            stopAndWait(cleaner.getExecutor(), "TTL workers");
         }
         if (storeStream != null) {
             // Cancelled queued scans must run their finally blocks to release iterators.
@@ -108,7 +104,18 @@ public void onApplicationEvent(ContextClosedEvent event) {
         if (storeStream != null) {
             storeStream.awaitScanCleanup();
         }
-        log.info("closed gRPC callbacks, scan and aggregate query workers");
+        if (cleaner != null) {
+            cleaner.awaitCleanup();
+        }
+        log.info("closed gRPC callbacks, scan, aggregate query and TTL workers");
+    }
+
+    private static void stopAndWait(ExecutorService executor, String name) {
+        if (executor == null) {
+            return;
+        }
+        executor.shutdownNow();
+        awaitWorkers(executor, name);
     }
 
     private static void awaitWorkers(ExecutorService executor, String name) {

hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/task/TTLCleaner.java

@@ -33,6 +33,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 
 import org.apache.commons.lang3.ArrayUtils;
@@ -97,6 +98,7 @@ public class TTLCleaner implements Runnable {
     @Autowired
     private HgStoreNodeService service;
     private final AtomicBoolean running = new AtomicBoolean(false);
+    private final AtomicReference<Throwable> cleanupFailure = new AtomicReference<>();
 
     private final AppConfig appConfig;
     private final AppConfig.JobConfig jobConfig;
@@ -285,11 +287,12 @@ private Runnable getTask(
             String table = t.getRight();
             TaskInfo taskInfo = counter.get(graph);
             ScanIterator scan = null;
+            RocksDBSession session = null;
             try {
                 Map<String, AtomicLong> graphCounter = taskInfo.getTableCounter();
                 TaskSubmitter submitter = taskInfo.getTaskSubmitter();
                 AtomicLong tableCounter = graphCounter.get(table);
-                RocksDBSession session = handler.getSession(id);
+                session = handler.getSession(id);
                 InnerKeyCreator keyCreator = handler.getKeyCreator();
                 SessionOperator op = session.sessionOp();
                 BiFunction<byte[], byte[], Boolean> judge = getJudge(graph, table);
@@ -301,7 +304,7 @@ private Runnable getTask(
                 LinkedList<ByteString> all = new LinkedList<>();
                 AtomicBoolean state = new AtomicBoolean(true);
                 AtomicLong partitionCounter = pc.get(id);
-                while (filter.hasNext() && state.get()) {
+                while (!Thread.currentThread().isInterrupted() && state.get() && filter.hasNext()) {
                     RocksDBSession.BackendColumn current = filter.next();
                     byte[] realKey =
                             Arrays.copyOfRange(current.name, 0, current.name.length - Short.BYTES);
@@ -315,7 +318,7 @@ private Runnable getTask(
                         all = new LinkedList<>();
                     }
                 }
-                if (all.size() > 0 && state.get()) {
+                if (!Thread.currentThread().isInterrupted() && all.size() > 0 && state.get()) {
                     submitter.submitClean(id, graph, table, all, state, tableCounter,
                                           partitionCounter);
                 }
@@ -326,14 +329,59 @@ private Runnable getTask(
                 String msg = String.format(s, id, graph, table);
                 log.error(msg, e);
             } finally {
-                latch.countDown();
-                if (scan != null) {
-                    scan.close();
+                try {
+                    if (scan != null) {
+                        scan.close();
+                    }
+                } catch (RuntimeException | Error failure) {
+                    recordCleanupFailure(failure);
+                } finally {
+                    try {
+                        if (session != null) {
+                            session.close();
+                        }
+                    } catch (RuntimeException | Error failure) {
+                        recordCleanupFailure(failure);
+                    } finally {
+                        latch.countDown();
+                    }
                 }
             }
         };
     }
 
+    private void recordCleanupFailure(Throwable failure) {
+        this.cleanupFailure.compareAndSet(null, failure);
+        Throwable first = this.cleanupFailure.get();
+        if (first != failure) {
+            first.addSuppressed(failure);
+        }
+        log.error("TTL native cleanup failed; database close stays blocked", failure);
+    }
+
+    /** Call after workers terminate: a failed native release must remain a shutdown blocker. */
+    public void awaitCleanup() {
+        boolean interrupted = false;
+        long nextLog = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        synchronized (this.cleanupFailure) {
+            while (this.cleanupFailure.get() != null) {
+                try {
+                    this.cleanupFailure.wait(5000L);
+                } catch (InterruptedException e) {
+                    interrupted = true;
+                }
+                if (System.nanoTime() - nextLog >= 0) {
+                    log.warn("TTL scan cleanup failed; database close stays blocked",
+                             this.cleanupFailure.get());
+                    nextLog = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+                }
+            }
+        }
+        if (interrupted) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
     public ScheduledFuture<?> getFuture() {
         return future;
     }