refactor(store): drain TTL native resources on shutdown (#3304)

Shutdown waits for TTL workers and their native cleanup after RPC/Scan ownership drains. TTL closes its scan and borrowed session lease in order, attempts both releases, and retains cleanup failures as shutdown blockers.
The distribution stop script reports timeouts explicitly. Native reference-count, close-failure, active-read and shutdown regressions ship here. A real nonempty TTL scan with interrupted Spring close verifies that the database stays alive while the reader is blocked, then checks iterator/session drain before bean destruction.

Author: imbajin

.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

@@ -400,10 +400,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;
     }