From b68d89d1e1b2bf7196b228bf744b6933c98c6e2e Mon Sep 17 00:00:00 2001 From: Himanshu Verma Date: Tue, 4 Aug 2026 01:07:18 +0530 Subject: [PATCH 1/5] fix(server): wait for GRAPH_CREATE event when creating graph on PD path The PD-backed createGraph fired GRAPH_CREATE without awaiting it, so the REST 200 could be written before ContextGremlinServer injected the graph into the Gremlin global bindings, and an immediate Gremlin/Cypher request to the creating server could fail with "Could not rebind [g]". createGraphLocal already waits via notifyAndWaitEvent; this applies the same call on the PD path. --- .../src/main/java/org/apache/hugegraph/core/GraphManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index f716285c67..4bb37181ce 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -1353,7 +1353,7 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, } // Let gremlin server and rest server context add graph - this.eventHub.notify(Events.GRAPH_CREATE, graph); + this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); if (init) { String schema = propConfig.getString( From 9a4ac02e7fc95f4a3974f02a53e57558bd0d9391 Mon Sep 17 00:00:00 2001 From: Himanshu Verma Date: Tue, 4 Aug 2026 13:30:32 +0530 Subject: [PATCH 2/5] fix(server): fail graph create when the local binding fails Waiting for the GRAPH_CREATE future was not enough to prove that the graph was actually registered: EventHub swallows every throwable raised by a listener and resolves the future with the number of listeners that returned normally, so a listener that blew up looked exactly like a successful one. The create now compares the notified count with the registered listener count and fails when a listener did not complete. The wait is also bounded now instead of blocking forever, and an InterruptedException restores the thread's interrupt status before the failure is reported. Ordering is fixed along with it. On the PD path the graph is bound in the local gremlin/rest server context before its config is written to meta and broadcast, so a failed binding cannot leave a graph behind in meta for the other servers to converge on. A binding failure now unregisters the graph locally and closes it, the same cleanup a failed backend init already does, rather than dropping data that other servers may have bound successfully. On the local path the notify moved inside the existing try, which now also unregisters the graph before dropping it, so a failed binding leaves no closed graph behind in the context. The drop path keeps the lenient behaviour: the data is already gone when the event fires, so failing the request cannot undo anything and the listener state may legitimately be absent already. --- .../apache/hugegraph/core/GraphManager.java | 99 +++++++++++++++++-- 1 file changed, 89 insertions(+), 10 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index 4bb37181ce..83f907a36d 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -34,8 +34,10 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -151,6 +153,13 @@ public final class GraphManager { public static final String DELIMITER = "-"; public static final String NAMESPACE_CREATE = "namespace_create"; private static final Logger LOG = Log.logger(GraphManager.class); + /* + * The graph create/drop listeners only do in-memory registrations (put the + * graph into the rest server context and into the gremlin server bindings), + * so they finish in microseconds on a healthy server. The bound is only a + * guard against a stuck or starved event worker. + */ + private static final long EVENT_WAIT_TIMEOUT = 30L; private KvStore kvStore; private final String cluster; @@ -1193,18 +1202,20 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { // Init graph and start it graph.create(this.graphsDir, this.globalNodeRoleInfo); + + // Let gremlin server and rest server add graph to context + this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); } catch (Throwable e) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); if (graph != null) { + // The create event may have added the graph to the context + this.graphs.remove(graph.spaceGraphName()); this.dropGraphLocal(graph); } throw e; } - // Let gremlin server and rest server add graph to context - this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); - return graph; } @@ -1342,19 +1353,37 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, graph.updateTime(timeStamp); String graphName = spaceGraphName(graphSpace, name); + this.graphs.put(graphName, graph); + + /* + * Let gremlin server and rest server context add graph before the + * graph is published, so that a failed local binding can't leave the + * graph behind in meta for the other servers to converge on + */ + try { + this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); + } catch (Throwable e) { + this.graphs.remove(graphName); + try { + graph.close(); + } catch (Exception e1) { + if (graph instanceof StandardHugeGraph) { + ((StandardHugeGraph) graph).clearSchedulerAndLock(); + } + } + HugeFactory.remove(graph); + throw e; + } + if (init) { this.creatingGraphs.add(graphName); this.metaManager.addGraphConfig(graphSpace, name, configs); this.metaManager.notifyGraphAdd(graphSpace, name); } - this.graphs.put(graphName, graph); if (!grpcThread) { this.metaManager.updateGraphSpaceConfig(graphSpace, gs); } - // Let gremlin server and rest server context add graph - this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); - if (init) { String schema = propConfig.getString( CoreOptions.SCHEMA_INIT_TEMPLATE.name()); @@ -1771,10 +1800,60 @@ private void listenMetaChanges() { this.metaManager.listenGraphClear(ConsumerWrapper.wrap(this::graphClearHandler)); } + /** + * Notify the listeners of `event` and wait for them to finish, failing if + * any listener did not complete successfully. + *

+ * EventHub swallows every throwable raised by a listener and resolves the + * future with the number of listeners that returned normally, so waiting + * alone doesn't prove that the graph was registered. Comparing the + * notified count with the registered listener count detects the swallowed + * failure and lets the caller fail instead of returning a graph that is + * missing from the rest/gremlin server context. + */ private void notifyAndWaitEvent(String event, HugeGraph graph) { - Future future = this.eventHub.notify(event, graph); + String graphName = graph.spaceGraphName(); + // Listeners of ANY_EVENT are notified too, so they count as expected + int expected = this.eventHub.listeners(event).size() + + this.eventHub.listeners(EventHub.ANY_EVENT).size(); + int notified; try { - future.get(); + Future future = this.eventHub.notify(event, graph); + notified = future.get(EVENT_WAIT_TIMEOUT, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new HugeException("Interrupted while waiting for event " + + "'%s' of graph '%s'", e, event, graphName); + } catch (TimeoutException e) { + throw new HugeException("Timeout(%ss) while waiting for event " + + "'%s' of graph '%s'", e, + EVENT_WAIT_TIMEOUT, event, graphName); + } catch (ExecutionException e) { + throw new HugeException("Failed to wait for event '%s' of " + + "graph '%s'", e, event, graphName); + } + + if (notified < expected) { + throw new HugeException("Only %s of %s listeners handled event " + + "'%s' of graph '%s' successfully", + notified, expected, event, graphName); + } + } + + /** + * Same bounded wait as notifyAndWaitEvent(), but a failed listener is only + * logged. Used by the drop path: the graph data is already deleted when + * the event is sent, so failing the request can't undo anything, and the + * listener state may legitimately be absent already. + */ + private void notifyAndWaitEventLenient(String event, HugeGraph graph) { + try { + Future future = this.eventHub.notify(event, graph); + future.get(EVENT_WAIT_TIMEOUT, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOG.warn("Interrupted when waiting for event execution: {}", + event, e); } catch (Throwable e) { LOG.warn("Error when waiting for event execution: {}", event, e); } @@ -1999,7 +2078,7 @@ public void dropGraphLocal(String name) { this.dropGraphLocal(graph); // Let gremlin server and rest server context remove graph - this.notifyAndWaitEvent(Events.GRAPH_DROP, graph); + this.notifyAndWaitEventLenient(Events.GRAPH_DROP, graph); } public void dropGraph(String graphSpace, String name, boolean clear) { From e260ec990b05ef8eed656d22247284d631dff646 Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 12 Aug 2026 00:37:56 +0800 Subject: [PATCH 3/5] fix(server): make graph binding atomic - dispatch graph lifecycle events synchronously - roll back partial Gremlin registrations on failure - cover synchronous listener failure counting --- .../org/apache/hugegraph/event/EventHub.java | 60 ++++++++++------ .../hugegraph/unit/event/EventHubTest.java | 21 ++++++ .../hugegraph/auth/ContextGremlinServer.java | 8 ++- .../apache/hugegraph/core/GraphManager.java | 72 +++++-------------- 4 files changed, 85 insertions(+), 76 deletions(-) diff --git a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java index fbdf460a96..22a0dab7ac 100644 --- a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java +++ b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java @@ -167,10 +167,29 @@ public Future notify(String event, @Nullable Object... args) { return this.notify(event, null, args); } + /** + * Notify all registered listeners in the current thread. + */ + public int notifySync(String event, @Nullable Object... args) { + ExtendableIterator all = this.eventListeners(event); + return this.notifyListeners(all, null, new Event(this, event, args)); + } + private Future notify(String event, EventListener ignoredListener, @Nullable Object... args) { - @SuppressWarnings("resource") + ExtendableIterator all = this.eventListeners(event); + if (!all.hasNext()) { + return CompletableFuture.completedFuture(0); + } + Event ev = new Event(this, event, args); + return executor().submit(() -> { + return this.notifyListeners(all, ignoredListener, ev); + }); + } + + @SuppressWarnings("resource") + private ExtendableIterator eventListeners(String event) { ExtendableIterator all = new ExtendableIterator<>(); List ls = this.listeners.get(event); @@ -181,31 +200,30 @@ private Future notify(String event, if (lsAny != null && !lsAny.isEmpty()) { all.extend(lsAny.iterator()); } + return all; + } + private int notifyListeners(ExtendableIterator all, + EventListener ignoredListener, Event event) { if (!all.hasNext()) { - return CompletableFuture.completedFuture(0); + return 0; } - Event ev = new Event(this, event, args); - - // The submit will catch params: `all`(Listeners) and `ev`(Event) - return executor().submit(() -> { - int count = 0; - // Notify all listeners, and ignore the results - while (all.hasNext()) { - EventListener listener = all.next(); - if (listener == ignoredListener) { - continue; - } - try { - listener.event(ev); - count++; - } catch (Throwable e) { - LOG.warn("Failed to handle event: {}", ev, e); - } + int count = 0; + // Notify all listeners, and ignore the results + while (all.hasNext()) { + EventListener listener = all.next(); + if (listener == ignoredListener) { + continue; } - return count; - }); + try { + listener.event(event); + count++; + } catch (Throwable e) { + LOG.warn("Failed to handle event: {}", event, e); + } + } + return count; } public Object call(String event, @Nullable Object... args) { diff --git a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java index 69472bc8e3..4682e660a5 100644 --- a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java +++ b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java @@ -388,6 +388,27 @@ public void testEventNotifyWithArg2() { Assert.assertEquals(1, count.get()); } + @Test + public void testEventNotifySync() { + final String notify = "event-notify-sync"; + AtomicInteger count = new AtomicInteger(); + + this.eventHub.listen(notify, event -> { + count.incrementAndGet(); + return null; + }); + this.eventHub.listen(notify, event -> { + throw new RuntimeException("fake exception"); + }); + this.eventHub.listen(EventHub.ANY_EVENT, event -> { + count.incrementAndGet(); + return null; + }); + + Assert.assertEquals(2, this.eventHub.notifySync(notify)); + Assert.assertEquals(2, count.get()); + } + @Test public void testNotifyExcept() throws Exception { final String notify = "event-notify"; diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java index 0f5881b1a5..493d5e3612 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java @@ -146,8 +146,12 @@ private void removeGraph(String name) { GremlinExecutor executor = this.getServerGremlinExecutor() .getGremlinExecutor(); try { - manager.removeGraph(name); - manager.removeTraversalSource(G_PREFIX + name); + if (manager.getGraph(name) != null) { + manager.removeGraph(name); + } + if (manager.getTraversalSource(G_PREFIX + name) != null) { + manager.removeTraversalSource(G_PREFIX + name); + } Whitebox.invoke(executor, "globalBindings", new Class[]{Object.class}, "remove", name); diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index 83f907a36d..e6878ee61a 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -34,10 +34,7 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -153,13 +150,6 @@ public final class GraphManager { public static final String DELIMITER = "-"; public static final String NAMESPACE_CREATE = "namespace_create"; private static final Logger LOG = Log.logger(GraphManager.class); - /* - * The graph create/drop listeners only do in-memory registrations (put the - * graph into the rest server context and into the gremlin server bindings), - * so they finish in microseconds on a healthy server. The bound is only a - * guard against a stuck or starved event worker. - */ - private static final long EVENT_WAIT_TIMEOUT = 30L; private KvStore kvStore; private final String cluster; @@ -1204,13 +1194,13 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { graph.create(this.graphsDir, this.globalNodeRoleInfo); // Let gremlin server and rest server add graph to context - this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); + this.notifyEvent(Events.GRAPH_CREATE, graph); } catch (Throwable e) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); if (graph != null) { // The create event may have added the graph to the context - this.graphs.remove(graph.spaceGraphName()); + this.graphs.remove(graph.spaceGraphName(), graph); this.dropGraphLocal(graph); } throw e; @@ -1361,9 +1351,10 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, * graph behind in meta for the other servers to converge on */ try { - this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); + this.notifyEvent(Events.GRAPH_CREATE, graph); } catch (Throwable e) { - this.graphs.remove(graphName); + this.notifyEventLenient(Events.GRAPH_DROP, graph); + this.graphs.remove(graphName, graph); try { graph.close(); } catch (Exception e1) { @@ -1783,7 +1774,7 @@ private void listenChanges() { LOG.debug("RestServer accepts event '{}'", event.name()); event.checkArgs(HugeGraph.class); HugeGraph graph = (HugeGraph) event.args()[0]; - this.graphs.remove(graph.spaceGraphName()); + this.graphs.remove(graph.spaceGraphName(), graph); return null; }); } @@ -1801,37 +1792,19 @@ private void listenMetaChanges() { } /** - * Notify the listeners of `event` and wait for them to finish, failing if - * any listener did not complete successfully. + * Notify the listeners of `event` synchronously, failing if any listener + * did not complete successfully. *

- * EventHub swallows every throwable raised by a listener and resolves the - * future with the number of listeners that returned normally, so waiting - * alone doesn't prove that the graph was registered. Comparing the - * notified count with the registered listener count detects the swallowed - * failure and lets the caller fail instead of returning a graph that is - * missing from the rest/gremlin server context. + * EventHub swallows every throwable raised by a listener and returns the + * number of listeners that completed normally. Comparing it with the + * registered listener count detects the swallowed failure. */ - private void notifyAndWaitEvent(String event, HugeGraph graph) { + private void notifyEvent(String event, HugeGraph graph) { String graphName = graph.spaceGraphName(); // Listeners of ANY_EVENT are notified too, so they count as expected int expected = this.eventHub.listeners(event).size() + this.eventHub.listeners(EventHub.ANY_EVENT).size(); - int notified; - try { - Future future = this.eventHub.notify(event, graph); - notified = future.get(EVENT_WAIT_TIMEOUT, TimeUnit.SECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new HugeException("Interrupted while waiting for event " + - "'%s' of graph '%s'", e, event, graphName); - } catch (TimeoutException e) { - throw new HugeException("Timeout(%ss) while waiting for event " + - "'%s' of graph '%s'", e, - EVENT_WAIT_TIMEOUT, event, graphName); - } catch (ExecutionException e) { - throw new HugeException("Failed to wait for event '%s' of " + - "graph '%s'", e, event, graphName); - } + int notified = this.eventHub.notifySync(event, graph); if (notified < expected) { throw new HugeException("Only %s of %s listeners handled event " + @@ -1841,21 +1814,14 @@ private void notifyAndWaitEvent(String event, HugeGraph graph) { } /** - * Same bounded wait as notifyAndWaitEvent(), but a failed listener is only - * logged. Used by the drop path: the graph data is already deleted when - * the event is sent, so failing the request can't undo anything, and the - * listener state may legitimately be absent already. + * Notify listeners synchronously, but keep listener failures non-fatal. + * Used by the drop and rollback paths, where cleanup must be best-effort. */ - private void notifyAndWaitEventLenient(String event, HugeGraph graph) { + private void notifyEventLenient(String event, HugeGraph graph) { try { - Future future = this.eventHub.notify(event, graph); - future.get(EVENT_WAIT_TIMEOUT, TimeUnit.SECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - LOG.warn("Interrupted when waiting for event execution: {}", - event, e); + this.eventHub.notifySync(event, graph); } catch (Throwable e) { - LOG.warn("Error when waiting for event execution: {}", event, e); + LOG.warn("Error when notifying event: {}", event, e); } } @@ -2078,7 +2044,7 @@ public void dropGraphLocal(String name) { this.dropGraphLocal(graph); // Let gremlin server and rest server context remove graph - this.notifyAndWaitEventLenient(Events.GRAPH_DROP, graph); + this.notifyEventLenient(Events.GRAPH_DROP, graph); } public void dropGraph(String graphSpace, String name, boolean clear) { From 5569400b2ede0a54f9ebbf35e9e9d1ab75cf9bd1 Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 12 Aug 2026 00:47:24 +0800 Subject: [PATCH 4/5] fix(server): complete graph rollback - clean local bindings after failed creation - preserve backend cleanup before graph unbinding - remove Gremlin bindings before closing graph --- .../hugegraph/auth/ContextGremlinServer.java | 14 ++++++---- .../apache/hugegraph/core/GraphManager.java | 26 ++++++++++++------- 2 files changed, 25 insertions(+), 15 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java index 493d5e3612..e859050132 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java @@ -75,7 +75,7 @@ private void listenChanges() { LOG.debug("GremlinServer accepts event '{}'", event.name()); event.checkArgs(HugeGraph.class); HugeGraph graph = (HugeGraph) event.args()[0]; - this.removeGraph(graph.spaceGraphName()); + this.removeGraph(graph); return null; }); } @@ -123,7 +123,7 @@ public void injectTraversalSource() { } } - private void injectGraph(HugeGraph graph) { + private synchronized void injectGraph(HugeGraph graph) { String name = graph.spaceGraphName(); GraphManager manager = this.getServerGremlinExecutor() .getGraphManager(); @@ -140,14 +140,15 @@ private void injectGraph(HugeGraph graph) { "put", name, graph); } - private void removeGraph(String name) { + private synchronized void removeGraph(HugeGraph graph) { + String name = graph.spaceGraphName(); GraphManager manager = this.getServerGremlinExecutor() .getGraphManager(); GremlinExecutor executor = this.getServerGremlinExecutor() .getGremlinExecutor(); try { - if (manager.getGraph(name) != null) { - manager.removeGraph(name); + if (manager.getGraph(name) != graph) { + return; } if (manager.getTraversalSource(G_PREFIX + name) != null) { manager.removeTraversalSource(G_PREFIX + name); @@ -155,6 +156,9 @@ private void removeGraph(String name) { Whitebox.invoke(executor, "globalBindings", new Class[]{Object.class}, "remove", name); + if (manager.getGraph(name) != null) { + manager.removeGraph(name); + } } catch (Exception e) { throw new HugeException("Failed to remove graph '%s' from " + "gremlin server context", e, name); diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index e6878ee61a..18e33379b5 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -1199,9 +1199,13 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); if (graph != null) { - // The create event may have added the graph to the context this.graphs.remove(graph.spaceGraphName(), graph); - this.dropGraphLocal(graph); + try { + this.dropGraphLocal(graph); + } finally { + // The create event may have partially registered the graph + this.notifyEventLenient(Events.GRAPH_DROP, graph); + } } throw e; } @@ -1210,14 +1214,16 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { } private void dropGraphLocal(HugeGraph graph) { - // Clear data and config files - graph.drop(); - - /* - * Will fill graph instance into HugeFactory.graphs after - * GraphFactory.open() succeed, remove it when the graph drops - */ - HugeFactory.remove(graph); + try { + // Clear data and config files + graph.drop(); + } finally { + /* + * Will fill graph instance into HugeFactory.graphs after + * GraphFactory.open() succeed, remove it when the graph drops + */ + HugeFactory.remove(graph); + } } public HugeGraph createGraph(String graphSpace, String name, String creator, From 92997d8931835653c369f07ddfba01bf37b1dc8a Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 12 Aug 2026 01:02:37 +0800 Subject: [PATCH 5/5] fix(server): count event outcomes atomically - report attempted and successful listeners together - validate graph creation from one listener snapshot - cover listener mutation during synchronous dispatch --- .../org/apache/hugegraph/event/EventHub.java | 42 +++++++++++++++---- .../hugegraph/unit/event/EventHubTest.java | 21 +++++++++- .../apache/hugegraph/core/GraphManager.java | 16 ++++--- 3 files changed, 61 insertions(+), 18 deletions(-) diff --git a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java index 22a0dab7ac..0787b783d6 100644 --- a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java +++ b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java @@ -39,6 +39,29 @@ public class EventHub { + public static final class NotifyResult { + + private final int attempted; + private final int succeeded; + + private NotifyResult(int attempted, int succeeded) { + this.attempted = attempted; + this.succeeded = succeeded; + } + + public int attempted() { + return this.attempted; + } + + public int succeeded() { + return this.succeeded; + } + + public boolean success() { + return this.attempted == this.succeeded; + } + } + private static final Logger LOG = Log.logger(EventHub.class); public static final String EVENT_WORKER = "event-worker-%d"; @@ -170,7 +193,7 @@ public Future notify(String event, @Nullable Object... args) { /** * Notify all registered listeners in the current thread. */ - public int notifySync(String event, @Nullable Object... args) { + public NotifyResult notifySync(String event, @Nullable Object... args) { ExtendableIterator all = this.eventListeners(event); return this.notifyListeners(all, null, new Event(this, event, args)); } @@ -184,7 +207,7 @@ private Future notify(String event, } Event ev = new Event(this, event, args); return executor().submit(() -> { - return this.notifyListeners(all, ignoredListener, ev); + return this.notifyListeners(all, ignoredListener, ev).succeeded(); }); } @@ -203,27 +226,30 @@ private ExtendableIterator eventListeners(String event) { return all; } - private int notifyListeners(ExtendableIterator all, - EventListener ignoredListener, Event event) { + private NotifyResult notifyListeners(ExtendableIterator all, + EventListener ignoredListener, + Event event) { if (!all.hasNext()) { - return 0; + return new NotifyResult(0, 0); } - int count = 0; + int attempted = 0; + int succeeded = 0; // Notify all listeners, and ignore the results while (all.hasNext()) { EventListener listener = all.next(); if (listener == ignoredListener) { continue; } + attempted++; try { listener.event(event); - count++; + succeeded++; } catch (Throwable e) { LOG.warn("Failed to handle event: {}", event, e); } } - return count; + return new NotifyResult(attempted, succeeded); } public Object call(String event, @Nullable Object... args) { diff --git a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java index 4682e660a5..43e4721e6c 100644 --- a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java +++ b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java @@ -31,6 +31,7 @@ import org.apache.hugegraph.unit.BaseUnitTest; import org.apache.hugegraph.event.Event; import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.event.EventHub.NotifyResult; import org.apache.hugegraph.event.EventListener; public class EventHubTest extends BaseUnitTest { @@ -405,10 +406,28 @@ public void testEventNotifySync() { return null; }); - Assert.assertEquals(2, this.eventHub.notifySync(notify)); + NotifyResult result = this.eventHub.notifySync(notify); + Assert.assertEquals(3, result.attempted()); + Assert.assertEquals(2, result.succeeded()); + Assert.assertFalse(result.success()); Assert.assertEquals(2, count.get()); } + @Test + public void testEventNotifySyncUsesSingleListenerSnapshot() { + final String notify = "event-notify-sync-snapshot"; + this.eventHub.listen(notify, event -> { + this.eventHub.listen(notify, ignored -> null); + return null; + }); + + NotifyResult result = this.eventHub.notifySync(notify); + + Assert.assertEquals(1, result.attempted()); + Assert.assertEquals(1, result.succeeded()); + Assert.assertTrue(result.success()); + } + @Test public void testNotifyExcept() throws Exception { final String notify = "event-notify"; diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index 18e33379b5..bd74b712ff 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -65,6 +65,7 @@ import org.apache.hugegraph.config.ServerOptions; import org.apache.hugegraph.config.TypedOption; import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.event.EventHub.NotifyResult; import org.apache.hugegraph.exception.ExistedException; import org.apache.hugegraph.exception.NotFoundException; import org.apache.hugegraph.exception.NotSupportException; @@ -1801,21 +1802,18 @@ private void listenMetaChanges() { * Notify the listeners of `event` synchronously, failing if any listener * did not complete successfully. *

- * EventHub swallows every throwable raised by a listener and returns the - * number of listeners that completed normally. Comparing it with the - * registered listener count detects the swallowed failure. + * EventHub swallows every throwable raised by a listener and reports the + * attempted and successful listeners from the same snapshot. */ private void notifyEvent(String event, HugeGraph graph) { String graphName = graph.spaceGraphName(); - // Listeners of ANY_EVENT are notified too, so they count as expected - int expected = this.eventHub.listeners(event).size() + - this.eventHub.listeners(EventHub.ANY_EVENT).size(); - int notified = this.eventHub.notifySync(event, graph); + NotifyResult result = this.eventHub.notifySync(event, graph); - if (notified < expected) { + if (!result.success()) { throw new HugeException("Only %s of %s listeners handled event " + "'%s' of graph '%s' successfully", - notified, expected, event, graphName); + result.succeeded(), result.attempted(), + event, graphName); } }