fix(server): bound final partition diagnostic

docker/README.md

@@ -241,6 +241,10 @@ shutdown through the launcher using TERM, which shuts down and reaps the JVM; th
 The startup REST check still uses `HG_SERVER_STARTUP_TIMEOUT_S`, and HStore partition checks still
 run after readiness. Shutdown interrupts a stalled readiness probe or pending partition check and
 reaps its polling processes.
+The partition poll and its final `/v1/partitions` diagnostic each have a separate
+`WAIT_PARTITION_TIMEOUT_S` budget (120 seconds by default). If the JVM exits while the
+final diagnostic is stalled, the entrypoint observes its exit after that request finishes
+or reaches this bound; it does not monitor the JVM immediately during the diagnostic.
 A startup timeout requests termination and returns 1 without waiting for an
 unresponsive JVM beyond the startup deadline; container termination cleans up remaining processes.
 Graceful shutdown requires the JVM to respond to TERM; forceful OOM and orchestrator shutdown testing

hugegraph-server/hugegraph-dist/docker/test/test-docker-entrypoint-process.sh

@@ -52,7 +52,9 @@ with tempfile.TemporaryDirectory(prefix='hugegraph-entrypoint-process-') as dire
                            ('timeout-ignore', 1), ('partition42', 42),
                            ('partitionTERM', 143), ('partitionINT', 130),
                            ('partitiongroupTERM', 143), ('readinessTERM', 143),
-                           ('readinessINT', 130), ('readinessgroupTERM', 143)]:
+                           ('readinessINT', 130), ('readinessgroupTERM', 143),
+                           ('partitionDiagnostic42', 42), ('partitionDiagnosticTERM', 143),
+                           ('partitionDiagnosticINT', 130), ('partitionDiagnosticgroupTERM', 143)]:
         work = pathlib.Path(directory) / mode
         for name in ['bin', 'tools', 'conf/graphs', 'docker', 'logs']:
             (work / name).mkdir(parents=True)
@@ -66,6 +68,29 @@ with tempfile.TemporaryDirectory(prefix='hugegraph-entrypoint-process-') as dire
             'restserver.url=http://127.0.0.1:1\ngremlinserver.url=http://127.0.0.1:2\n')
         listener = None
         accepted, stop_http = threading.Event(), threading.Event()
+        if mode.startswith('partitionDiagnostic'):
+            listener = socket.socket()
+            listener.bind(('127.0.0.1', 0))
+            listener.listen(2)
+            port = listener.getsockname()[1]
+
+            def hold_partition_diagnostic():
+                # Real curl first receives positive readiness, then stalls on
+                # the distinct final diagnostic endpoint.
+                for _ in range(2):
+                    connection, _ = listener.accept()
+                    request = connection.recv(8192)
+                    if b' /v1/partitions ' in request:
+                        accepted.set()
+                        stop_http.wait(10)
+                        connection.close()
+                        return
+                    body = b'{"partitionCount":1}'
+                    response = b'HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n' % len(body)
+                    connection.sendall(response + body)
+                    connection.close()
+
+            threading.Thread(target=hold_partition_diagnostic, daemon=True).start()
         if mode.startswith('readiness'):
             listener = socket.socket()
             listener.bind(('127.0.0.1', 0))
@@ -95,6 +120,15 @@ with tempfile.TemporaryDirectory(prefix='hugegraph-entrypoint-process-') as dire
 case "$TEST_CHILD_MODE" in
     early* | timeout*) printf '000'; exit 28 ;;
 esac
+if [[ "$1" != '-I' && "$TEST_CHILD_MODE" == partitionDiagnostic* ]]; then
+    if [[ "$1" == '-v' ]]; then
+        echo "$$" > ./partition-curl.pid
+        echo "$PPID" > ./partition-timeout.pid
+        ps -o ppid= -p "$PPID" | tr -d ' ' > ./partition-wrapper.pid
+        touch ./partition-ran
+    fi
+    exec "$TEST_REAL_CURL" "$@"
+fi
 if [[ "$TEST_CHILD_MODE" == readiness* ]]; then
     ancestor=$$
     for index in 1 2 3 4; do
@@ -148,6 +182,9 @@ sys.exit(int((work / 'child-exit').read_text()))
                    TEST_REAL_CURL=shutil.which('curl'), NO_PROXY='127.0.0.1', no_proxy='127.0.0.1')
         if mode.startswith('readiness'):
             env['HG_SERVER_STARTUP_TIMEOUT_S'] = '60'
+        if mode.startswith('partitionDiagnostic'):
+            env['STORE_REST'] = '127.0.0.1:%d' % port
+            env['WAIT_PARTITION_TIMEOUT_S'] = '1' if mode.endswith('42') else '10'
         started = time.monotonic()
         with (work / 'output').open('wb') as output:
             process = subprocess.Popen(['bash', './entrypoint.sh'], cwd=work, env=env,
@@ -169,23 +206,27 @@ sys.exit(int((work / 'child-exit').read_text()))
                     await_file(work / 'ready-probe', process)
                     if mode.startswith('partition'):
                         await_file(work / 'partition-ran', process)
-                    if mode in ['TERM', 'INT', 'partitionTERM', 'partitionINT']:
+                    if mode.startswith('partitionDiagnostic'):
+                        assert accepted.wait(5), 'final real curl did not reach the stalled endpoint'
+                    if mode in ['TERM', 'INT', 'partitionTERM', 'partitionINT',
+                                'partitionDiagnosticTERM', 'partitionDiagnosticINT']:
                         signum = signal.SIGINT if mode.endswith('INT') else signal.SIGTERM
                         os.kill(process.pid, signum)
-                    elif mode in ['groupTERM', 'partitiongroupTERM']:
+                    elif mode in ['groupTERM', 'partitiongroupTERM', 'partitionDiagnosticgroupTERM']:
                         os.killpg(process.pid, signal.SIGTERM)
                     elif mode == 'kill137':
                         os.kill(int((work / 'bin/pid').read_text()), signal.SIGKILL)
                     else:
-                        status = 42 if mode == 'partition42' else int(mode[4:])
+                        status = 42 if mode in ['partition42', 'partitionDiagnostic42'] else int(mode[4:])
                         (work / 'child-exit').write_text(str(status))
                 actual = process.wait(timeout=5)
                 assert actual == expected, (mode, expected, actual, (work / 'output').read_text())
                 if mode.startswith('timeout'):
                     assert time.monotonic() - started < 5, 'startup cleanup exceeded the deadline'
                 if mode in ['TERM', 'INT', 'groupTERM', 'partitionTERM', 'partitionINT',
                             'partitiongroupTERM', 'readinessTERM', 'readinessINT',
-                            'readinessgroupTERM']:
+                            'readinessgroupTERM', 'partitionDiagnosticTERM',
+                            'partitionDiagnosticINT', 'partitionDiagnosticgroupTERM']:
                     assert (work / 'term-observed').exists(), 'TERM did not reach the JVM substitute'
                 if not mode.startswith('timeout'):
                     owned = ([work / 'bin/pid'] + list(work.glob('partition-*.pid')) +
@@ -198,7 +239,7 @@ sys.exit(int((work / 'child-exit').read_text()))
                             pass
                         else:
                             raise AssertionError(str(pid_file) + ' survived entrypoint exit')
-                if mode == 'partition42':
+                if mode in ['partition42', 'partitionDiagnostic42']:
                     assert 'WARN: partitions not assigned yet' in (work / 'output').read_text()
                 print('PASS: ' + mode + ' -> ' + str(actual))
             finally:
@@ -226,3 +267,6 @@ sys.exit(int((work / 'child-exit').read_text()))
                     stop_http.set()
                     listener.close()
 PY
+
+# This process fixture is already invoked by the Docker/Server CI entrypoint tests.
+bash "$(dirname "${BASH_SOURCE[0]}")/test-wait-partition.sh"

hugegraph-server/hugegraph-dist/docker/test/test-wait-partition.sh

@@ -0,0 +1,123 @@
+#!/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.
+#
+set -euo pipefail
+
+REPO_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd)
+SCRIPT=${1:-"$REPO_ROOT/hugegraph-server/hugegraph-dist/src/assembly/static/bin/wait-partition.sh"}
+# One real loopback socket and GNU timeout; no JVM, database or Docker.
+python3 - "$SCRIPT" <<'PY'
+import os
+import pathlib
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+script = pathlib.Path(sys.argv[1]).resolve()
+with tempfile.TemporaryDirectory(prefix='hugegraph-partition-diagnostic-') as directory:
+    for mode, expected in [('healthy', 0), ('stalled', 124), ('TERM', 143), ('groupTERM', 143)]:
+        work = pathlib.Path(directory) / mode
+        work.mkdir()
+        tools = work / 'tools'
+        tools.mkdir()
+        (tools / 'curl').write_text('''#!/bin/bash
+if [[ "$1" == '-v' ]]; then
+    echo "$$" > ./curl.pid
+    echo "$PPID" > ./timeout.pid
+fi
+exec "$REAL_CURL" "$@"
+''')
+        (tools / 'curl').chmod(0o755)
+        listener = socket.socket()
+        listener.bind(('127.0.0.1', 0))
+        listener.listen(2)
+        accepted, stop = threading.Event(), threading.Event()
+
+        def serve():
+            for _ in range(2):
+                connection, _ = listener.accept()
+                request = connection.recv(8192)
+                if b' /v1/partitions ' in request:
+                    accepted.set()
+                    if mode != 'healthy':
+                        stop.wait(10)
+                        connection.close()
+                        return
+                    body = b'{"partitions":[]}'
+                else:
+                    body = b'{"partitionCount":1}'
+                response = b'HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n' % len(body)
+                connection.sendall(response + body)
+                connection.close()
+
+        threading.Thread(target=serve, daemon=True).start()
+        env = dict(os.environ, STORE_REST='127.0.0.1:%d' % listener.getsockname()[1],
+                   WAIT_PARTITION_TIMEOUT_S='1' if mode == 'stalled' else '10',
+                   PATH=str(tools) + ':' + os.environ['PATH'], REAL_CURL=shutil.which('curl'),
+                   NO_PROXY='127.0.0.1', no_proxy='127.0.0.1')
+        started = time.monotonic()
+        with (work / 'output').open('wb') as output:
+            process = subprocess.Popen(['bash', str(script)], cwd=work, env=env,
+                                       stdout=output, stderr=subprocess.STDOUT, start_new_session=True)
+            try:
+                assert accepted.wait(3), 'final diagnostic did not reach the real socket'
+                if mode == 'TERM':
+                    os.kill(process.pid, signal.SIGTERM)
+                elif mode == 'groupTERM':
+                    os.killpg(process.pid, signal.SIGTERM)
+                actual = process.wait(timeout=4)
+                assert actual == expected, (mode, expected, actual, (work / 'output').read_text())
+                for pid_file in [work / 'curl.pid', work / 'timeout.pid']:
+                    owned = int(pid_file.read_text())
+                    try:
+                        os.kill(owned, 0)
+                    except ProcessLookupError:
+                        pass
+                    else:
+                        raise AssertionError(str(pid_file) + ' survived helper exit')
+                print('PASS: final diagnostic %s -> %d (%.2fs); curl/timeout reaped' %
+                      (mode, actual, time.monotonic() - started))
+            finally:
+                # A regression may leave GNU timeout in its own process group.
+                # Only this fixture's recorded process IDs/groups are touched.
+                for pid_file in [work / 'timeout.pid', work / 'curl.pid']:
+                    if pid_file.exists():
+                        owned = int(pid_file.read_text())
+                        try:
+                            os.killpg(owned, signal.SIGKILL)
+                        except ProcessLookupError:
+                            pass
+                        try:
+                            os.kill(owned, signal.SIGKILL)
+                        except ProcessLookupError:
+                            pass
+                try:
+                    process.wait(timeout=0.5)
+                except subprocess.TimeoutExpired:
+                    try:
+                        os.killpg(process.pid, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+                    process.wait(timeout=2)
+                stop.set()
+                listener.close()
+PY

hugegraph-server/hugegraph-dist/src/assembly/static/bin/wait-partition.sh

@@ -51,7 +51,8 @@ CHECK_PID=""
 echo "Partitions detected:"
 URL="http://${STORE_REST}/v1/partitions"
 echo "$URL"
-curl -v "$URL" &
+# The final diagnostic needs its own bound after partition polling succeeds.
+timeout "${WAIT_PARTITION_TIMEOUT_S:-120}s" curl -v "$URL" &
 CHECK_PID=$!
 STATUS=0
 wait "$CHECK_PID" || STATUS=$?