From d7369e73a4da18aed212e06060a1e6aeaab3f3e2 Mon Sep 17 00:00:00 2001 From: orion Date: Sat, 31 Oct 2026 05:54:40 +0000 Subject: [PATCH] fix(store): expose process lifecycle separately from readiness --- hugegraph-store/docs/deployment-guide.md | 12 ++ .../controller/HgStoreStatusController.java | 12 ++ .../node/listener/StoreProcessLifecycle.java | 78 ++++++++ .../store/service/ServerSuiteTest.java | 1 + .../service/StoreProcessLifecycleTest.java | 189 ++++++++++++++++++ 5 files changed, 292 insertions(+) create mode 100644 hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/listener/StoreProcessLifecycle.java create mode 100644 hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/StoreProcessLifecycleTest.java diff --git a/hugegraph-store/docs/deployment-guide.md b/hugegraph-store/docs/deployment-guide.md index 1e31c92137..c0440f256f 100644 --- a/hugegraph-store/docs/deployment-guide.md +++ b/hugegraph-store/docs/deployment-guide.md @@ -697,6 +697,18 @@ tail -f logs/hugegraph-store.log curl http://localhost:8520/v1/health ``` +On Store versions that provide it, `GET /-/lifecycle` reports the process +startup lifecycle as `{"status":0,"text":"STARTING"}`, `STARTED`, +`STOPPING`, or `UNAVAILABLE`. `STARTED` means the application reached Spring +Ready and the gRPC listener started and has not shut down. It does **not** +establish PD registration, partition recovery, Raft catch-up, or graph read/write +readiness. `STOPPING` takes precedence once context shutdown begins; +`UNAVAILABLE` means the gRPC server has shut down before that point. An older +Store may return 404, which means this signal is unavailable, not that the Store +is down. The existing `/-/state` operator-controlled flag is unchanged and must +not be inferred from this lifecycle value. Continue to assess data readiness +separately; do not use this endpoint as a rolling-update safety gate. + **Verify Store registration with PD**: ```bash diff --git a/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/controller/HgStoreStatusController.java b/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/controller/HgStoreStatusController.java index 86a6830795..b073dbb929 100644 --- a/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/controller/HgStoreStatusController.java +++ b/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/controller/HgStoreStatusController.java @@ -23,6 +23,7 @@ import org.apache.hugegraph.store.node.entry.RestResult; import org.apache.hugegraph.store.node.grpc.HgStoreNodeState; import org.apache.hugegraph.store.node.grpc.HgStoreStreamImpl; +import org.apache.hugegraph.store.node.listener.StoreProcessLifecycle; import org.apache.hugegraph.store.node.model.HgNodeStatus; import org.apache.hugegraph.store.node.task.TTLCleaner; import org.springframework.beans.factory.annotation.Autowired; @@ -41,6 +42,12 @@ @RestController public class HgStoreStatusController { + private final StoreProcessLifecycle processLifecycle; + + public HgStoreStatusController(StoreProcessLifecycle processLifecycle) { + this.processLifecycle = processLifecycle; + } + @Autowired HgStoreStreamImpl streamImpl; @Autowired @@ -57,6 +64,11 @@ public HgNodeStatus getState() { return new HgNodeStatus(0, HgStoreNodeState.getState().name()); } + @GetMapping("/-/lifecycle") + public HgNodeStatus getLifecycle() { + return new HgNodeStatus(0, this.processLifecycle.getState()); + } + @PutMapping("/-/state") public HgNodeStatus setState(@RequestParam(value = "name") String name) { diff --git a/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/listener/StoreProcessLifecycle.java b/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/listener/StoreProcessLifecycle.java new file mode 100644 index 0000000000..329634d171 --- /dev/null +++ b/hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/listener/StoreProcessLifecycle.java @@ -0,0 +1,78 @@ +/* + * 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. + */ + +package org.apache.hugegraph.store.node.listener; + +import org.lognet.springboot.grpc.context.GRpcServerInitializedEvent; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.ApplicationContext; +import org.springframework.context.event.ContextClosedEvent; +import org.springframework.context.event.EventListener; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import io.grpc.Server; + +/** + * Reports process startup only. This is not a shard or data readiness check. + */ +@Component +public class StoreProcessLifecycle { + + private final ApplicationContext context; + private boolean applicationReady; + private boolean stopping; + private Server grpcServer; + + public StoreProcessLifecycle(ApplicationContext context) { + this.context = context; + } + + @EventListener + public synchronized void onGrpcStarted(GRpcServerInitializedEvent event) { + if (event.getApplicationContext() == this.context && !this.stopping) { + this.grpcServer = event.getServer(); + } + } + + @EventListener + public synchronized void onApplicationReady(ApplicationReadyEvent event) { + if (event.getApplicationContext() == this.context && !this.stopping) { + this.applicationReady = true; + } + } + + @EventListener + @Order(Ordered.HIGHEST_PRECEDENCE) + public synchronized void onContextClosed(ContextClosedEvent event) { + if (event.getApplicationContext() == this.context) { + this.stopping = true; + } + } + + public synchronized String getState() { + if (this.stopping) { + return "STOPPING"; + } + if (this.grpcServer != null && + (this.grpcServer.isShutdown() || this.grpcServer.isTerminated())) { + return "UNAVAILABLE"; + } + return this.applicationReady && this.grpcServer != null ? "STARTED" : "STARTING"; + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/ServerSuiteTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/ServerSuiteTest.java index 763b7c47ea..7bfc1e750b 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/ServerSuiteTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/ServerSuiteTest.java @@ -38,6 +38,7 @@ OrderedMultiPartitionIteratorTest.class, ScanContextCleanupTest.class, ContextClosedListenerTest.class, + StoreProcessLifecycleTest.class, TTLCleanerCleanupTest.class, GrpcShutdownBarrierTest.class, PartitionEngineShutdownTest.class, diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/StoreProcessLifecycleTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/StoreProcessLifecycleTest.java new file mode 100644 index 0000000000..f8c51f171c --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/StoreProcessLifecycleTest.java @@ -0,0 +1,189 @@ +/* + * 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. + */ + +package org.apache.hugegraph.store.service; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.hugegraph.store.node.controller.HgStoreStatusController; +import org.apache.hugegraph.store.node.grpc.HgStoreNodeState; +import org.apache.hugegraph.store.node.listener.StoreProcessLifecycle; +import org.junit.Test; +import org.lognet.springboot.grpc.context.GRpcServerInitializedEvent; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.ApplicationListener; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.event.ContextClosedEvent; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.grpc.Server; + +public class StoreProcessLifecycleTest { + + @Test + public void endpointReportsOnlyThisProcessLifecycle() throws Exception { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + context.register(StoreProcessLifecycle.class); + context.refresh(); + StoreProcessLifecycle lifecycle = context.getBean(StoreProcessLifecycle.class); + HgStoreStatusController controller = new HgStoreStatusController(lifecycle); + ObjectMapper mapper = new ObjectMapper(); + Server server = mock(Server.class); + HgStoreNodeState.goStarting(); + try { + JsonNode initial = mapper.valueToTree(controller.getLifecycle()); + assertEquals(0, initial.get("status").asInt()); + assertEquals("STARTING", initial.get("text").asText()); + context.publishEvent(new GRpcServerInitializedEvent(context, server)); + assertEquals("STARTING", controller.getLifecycle().getText()); + context.publishEvent(new ApplicationReadyEvent(new SpringApplication(), + new String[0], context)); + assertEquals("STARTED", mapper.valueToTree(controller.getLifecycle()) + .get("text").asText()); + // The existing state endpoint remains independent of process startup. + assertEquals("STARTING", controller.getState().getText()); + HgStoreNodeState.goOnline(); + assertEquals("ONLINE", controller.getState().getText()); + assertEquals("STARTED", controller.getLifecycle().getText()); + when(server.isShutdown()).thenReturn(true); + assertEquals("UNAVAILABLE", controller.getLifecycle().getText()); + } finally { + HgStoreNodeState.goStarting(); + context.close(); + } + assertEquals("STOPPING", controller.getLifecycle().getText()); + // Late startup callbacks cannot resurrect a stopping process. + lifecycle.onGrpcStarted(new GRpcServerInitializedEvent(context, server)); + lifecycle.onApplicationReady(new ApplicationReadyEvent(new SpringApplication(), + new String[0], context)); + assertEquals("STOPPING", controller.getLifecycle().getText()); + } + + @Test + public void ignoresEventsFromAnotherContext() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + AnnotationConfigApplicationContext other = new AnnotationConfigApplicationContext(); + try { + context.register(StoreProcessLifecycle.class); + context.refresh(); + other.refresh(); + StoreProcessLifecycle lifecycle = context.getBean(StoreProcessLifecycle.class); + lifecycle.onGrpcStarted(new GRpcServerInitializedEvent(other, mock(Server.class))); + lifecycle.onApplicationReady(new ApplicationReadyEvent(new SpringApplication(), + new String[0], other)); + assertEquals("STARTING", lifecycle.getState()); + other.close(); + assertEquals("STARTING", lifecycle.getState()); + } finally { + other.close(); + context.close(); + } + } + + @Test + public void startedRequiresBothSignalsInEitherOrder() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + try { + StoreProcessLifecycle lifecycle = new StoreProcessLifecycle(context); + Server server = mock(Server.class); + lifecycle.onApplicationReady(new ApplicationReadyEvent(new SpringApplication(), + new String[0], context)); + assertEquals("STARTING", lifecycle.getState()); + lifecycle.onGrpcStarted(new GRpcServerInitializedEvent(context, server)); + assertEquals("STARTED", lifecycle.getState()); + when(server.isTerminated()).thenReturn(true); + assertEquals("UNAVAILABLE", lifecycle.getState()); + } finally { + context.close(); + } + } + + @Test + public void concurrentStartupCannotOverrideShutdown() throws Exception { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + for (int i = 0; i < 100; i++) { + StoreProcessLifecycle lifecycle = new StoreProcessLifecycle(context); + CountDownLatch start = new CountDownLatch(1); + Future ready = executor.submit(() -> { + start.await(); + lifecycle.onGrpcStarted(new GRpcServerInitializedEvent(context, mock(Server.class))); + lifecycle.onApplicationReady(new ApplicationReadyEvent(new SpringApplication(), + new String[0], context)); + return null; + }); + Future stop = executor.submit(() -> { + start.await(); + lifecycle.onContextClosed(new ContextClosedEvent(context)); + return null; + }); + start.countDown(); + ready.get(); + stop.get(); + assertEquals("STOPPING", lifecycle.getState()); + } + } finally { + executor.shutdownNow(); + context.close(); + } + } + + @Test(timeout = 5000) + public void stoppingIsVisibleBeforeOtherShutdownListenersFinish() throws Exception { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + CountDownLatch draining = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + context.register(StoreProcessLifecycle.class); + context.addApplicationListener((ApplicationListener) event -> { + draining.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + context.refresh(); + StoreProcessLifecycle lifecycle = context.getBean(StoreProcessLifecycle.class); + lifecycle.onGrpcStarted(new GRpcServerInitializedEvent(context, mock(Server.class))); + lifecycle.onApplicationReady(new ApplicationReadyEvent(new SpringApplication(), + new String[0], context)); + Future close = executor.submit(context::close); + assertTrue(draining.await(2, TimeUnit.SECONDS)); + assertEquals("STOPPING", lifecycle.getState()); + release.countDown(); + close.get(2, TimeUnit.SECONDS); + } finally { + release.countDown(); + executor.shutdownNow(); + context.close(); + } + } +}