fix(server): ref-count cache event listeners

- Replace the per-transaction cache listener owner flag with a per-graph
  CacheListenerHolder shared by graph and schema cache transactions.

- The holder keeps the EventHub listener registered while any transaction for the
  graph is alive, and unregisters/removes it only when the last transaction
  releases it. The registry update, ref-count decrement, and hub unlisten now run
  inside ConcurrentMap.compute() to avoid owner-closes-first invalidation gaps.

  Also add graph/schema regression coverage for owner-first close and last-close
  cleanup, including graph close/reopen handling for stale EventHub holders.

Author: dpol1

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CacheListenerHolder.java

@@ -0,0 +1,39 @@
+/*
+ * 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.backend.cache;
+
+import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventListener;
+
+/*
+ * Listener lifetime must cover all active transactions for the graph.
+ * The holder is removed from the registry and unregistered from EventHub
+ * only when the last transaction releases it.
+ */
+final class CacheListenerHolder {
+
+    final EventListener listener;
+    final EventHub hub;
+    int refCount;
+
+    CacheListenerHolder(EventListener listener, EventHub hub) {
+        this.listener = listener;
+        this.hub = hub;
+        this.refCount = 1;
+    }
+}

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java

@@ -24,6 +24,8 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import org.apache.hugegraph.HugeGraphParams;
 import org.apache.hugegraph.backend.cache.CachedBackendStore.QueryId;
@@ -60,12 +62,20 @@ public final class CachedGraphTransaction extends GraphTransaction {
     private static final long AVG_VERTEX_ENTRY_SIZE = 40L;
     private static final long AVG_EDGE_ENTRY_SIZE = 100L;
 
+    /*
+     * Listener lifetime must cover all active transactions for the graph.
+     * The holder is removed from the registry and unregistered from EventHub
+     * only when the last transaction releases it.
+     */
+    private static final ConcurrentMap<String, CacheListenerHolder>
+            graphCacheEventListeners = new ConcurrentHashMap<>();
+
     private final Cache<Id, Object> verticesCache;
     private final Cache<Id, Object> edgesCache;
 
     private EventListener storeEventListener;
     private EventListener cacheEventListener;
-    private boolean registeredCacheEventListener;
+    private CacheListenerHolder holder;
 
     public CachedGraphTransaction(HugeGraphParams graph, BackendStore store) {
         super(graph, store);
@@ -186,28 +196,44 @@ private void listenChanges() {
             return false;
         };
         EventHub graphEventHub = this.params().graphEventHub();
-        EventListener previous =
-                graphCacheEventListeners.putIfAbsent(
-                        this.params().spaceGraphName(), listener);
-        if (previous == null) {
-            this.cacheEventListener = listener;
-            this.registeredCacheEventListener = true;
-            graphEventHub.listen(Events.CACHE, this.cacheEventListener);
-        } else {
-            this.cacheEventListener = previous;
-            this.registeredCacheEventListener = false;
-        }
+        String graphName = this.params().spaceGraphName();
+        CacheListenerHolder acquired = graphCacheEventListeners.compute(
+                graphName, (key, existing) -> {
+                    if (existing == null || existing.hub != graphEventHub) {
+                        // Graph close/reopen creates a new EventHub for the
+                        // same graph name; replace the stale holder. Old
+                        // transactions skip decrement via identity check.
+                        if (existing != null) {
+                            existing.hub.unlisten(Events.CACHE,
+                                                  existing.listener);
+                        }
+                        graphEventHub.listen(Events.CACHE, listener);
+                        return new CacheListenerHolder(listener, graphEventHub);
+                    }
+                    existing.refCount++;
+                    return existing;
+                });
+        this.holder = acquired;
+        this.cacheEventListener = acquired.listener;
     }
 
     private void unlistenChanges() {
         String graphName = this.params().spaceGraphName();
-        if (this.registeredCacheEventListener) {
-            EventHub graphEventHub = this.params().graphEventHub();
-            if (graphCacheEventListeners.remove(graphName,
-                                                this.cacheEventListener)) {
-                graphEventHub.unlisten(Events.CACHE, this.cacheEventListener);
-            }
-            this.registeredCacheEventListener = false;
+        CacheListenerHolder ours = this.holder;
+        if (ours != null) {
+            graphCacheEventListeners.compute(graphName, (key, existing) -> {
+                if (existing == null || existing != ours) {
+                    return existing;
+                }
+                existing.refCount--;
+                if (existing.refCount == 0) {
+                    existing.hub.unlisten(Events.CACHE, existing.listener);
+                    return null;
+                }
+                return existing;
+            });
+            this.holder = null;
+            this.cacheEventListener = null;
         }
         if (storeEventListenStatus.remove(graphName) != null) {
             this.store().provider().unlisten(this.storeEventListener);

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java

@@ -43,7 +43,12 @@
 
 public final class CachedSchemaTransaction extends SchemaTransaction {
 
-    private static final ConcurrentMap<String, EventListener>
+    /*
+     * Listener lifetime must cover all active transactions for the graph.
+     * The holder is removed from the registry and unregistered from EventHub
+     * only when the last transaction releases it.
+     */
+    private static final ConcurrentMap<String, CacheListenerHolder>
             SCHEMA_CACHE_EVENT_LISTENERS = new ConcurrentHashMap<>();
 
     private final Cache<Id, Object> idCache;
@@ -53,7 +58,7 @@ public final class CachedSchemaTransaction extends SchemaTransaction {
 
     private EventListener storeEventListener;
     private EventListener cacheEventListener;
-    private boolean registeredCacheEventListener;
+    private CacheListenerHolder holder;
 
     public CachedSchemaTransaction(HugeGraphParams graph, BackendStore store) {
         super(graph, store);
@@ -138,29 +143,49 @@ private void listenChanges() {
         };
         EventHub schemaEventHub = this.params().schemaEventHub();
         String graph = this.params().spaceGraphName();
-        EventListener previous =
-                SCHEMA_CACHE_EVENT_LISTENERS.putIfAbsent(graph, listener);
-        if (previous == null) {
-            this.cacheEventListener = listener;
-            this.registeredCacheEventListener = true;
-            schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
-        } else {
-            this.cacheEventListener = previous;
-            this.registeredCacheEventListener = false;
-        }
+        CacheListenerHolder acquired = SCHEMA_CACHE_EVENT_LISTENERS.compute(
+                graph, (key, existing) -> {
+                    if (existing == null || existing.hub != schemaEventHub) {
+                        // Graph close/reopen creates a new EventHub for the
+                        // same graph name; replace the stale holder. Old
+                        // transactions skip decrement via identity check.
+                        if (existing != null) {
+                            existing.hub.unlisten(Events.CACHE,
+                                                  existing.listener);
+                        }
+                        schemaEventHub.listen(Events.CACHE, listener);
+                        return new CacheListenerHolder(listener,
+                                                       schemaEventHub);
+                    }
+                    existing.refCount++;
+                    return existing;
+                });
+        this.holder = acquired;
+        this.cacheEventListener = acquired.listener;
     }
 
     private void unlistenChanges() {
         // Unlisten store event
         this.store().provider().unlisten(this.storeEventListener);
 
         // Unlisten cache event
-        EventHub schemaEventHub = this.params().schemaEventHub();
-        if (this.registeredCacheEventListener) {
-            schemaEventHub.unlisten(Events.CACHE, this.cacheEventListener);
-            SCHEMA_CACHE_EVENT_LISTENERS.remove(this.params().spaceGraphName(),
-                                               this.cacheEventListener);
-            this.registeredCacheEventListener = false;
+        CacheListenerHolder ours = this.holder;
+        if (ours != null) {
+            SCHEMA_CACHE_EVENT_LISTENERS.compute(
+                    this.params().spaceGraphName(), (key, existing) -> {
+                        if (existing == null || existing != ours) {
+                            return existing;
+                        }
+                        existing.refCount--;
+                        if (existing.refCount == 0) {
+                            existing.hub.unlisten(Events.CACHE,
+                                                  existing.listener);
+                            return null;
+                        }
+                        return existing;
+                    });
+            this.holder = null;
+            this.cacheEventListener = null;
         }
     }
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java

@@ -60,7 +60,6 @@
 import org.apache.hugegraph.backend.store.BackendStore;
 import org.apache.hugegraph.config.CoreOptions;
 import org.apache.hugegraph.config.HugeConfig;
-import org.apache.hugegraph.event.EventListener;
 import org.apache.hugegraph.exception.LimitExceedException;
 import org.apache.hugegraph.exception.NotFoundException;
 import org.apache.hugegraph.iterator.BatchMapperIterator;
@@ -141,8 +140,6 @@ public class GraphTransaction extends IndexableTransaction {
 
     private final int verticesCapacity;
     private final int edgesCapacity;
-    protected static final ConcurrentHashMap<String, EventListener>
-            graphCacheEventListeners = new ConcurrentHashMap<>();
     protected static final ConcurrentHashMap<String, Boolean> storeEventListenStatus =
             new ConcurrentHashMap<>();