3535import org .apache .hugegraph .backend .query .QueryResults ;
3636import org .apache .hugegraph .backend .store .BackendMutation ;
3737import org .apache .hugegraph .backend .store .BackendStore ;
38+ import org .apache .hugegraph .backend .store .BackendStoreProvider ;
3839import org .apache .hugegraph .backend .store .ram .RamTable ;
3940import org .apache .hugegraph .backend .tx .GraphTransaction ;
4041import org .apache .hugegraph .config .CoreOptions ;
@@ -70,12 +71,22 @@ public final class CachedGraphTransaction extends GraphTransaction {
7071 private static final ConcurrentMap <String , CacheListenerHolder >
7172 GRAPH_CACHE_EVENT_LISTENERS = new ConcurrentHashMap <>();
7273
74+ /*
75+ * Same ref-counted lifecycle for the store event listener registered
76+ * on the BackendStoreProvider; see StoreListenerHolder.
77+ *
78+ * Replaces the removed protected static storeEventListenStatus field
79+ * that previously tracked store-listen state on GraphTransaction.
80+ */
81+ private static final ConcurrentMap <String , StoreListenerHolder >
82+ STORE_EVENT_LISTENERS = new ConcurrentHashMap <>();
83+
7384 private final Cache <Id , Object > verticesCache ;
7485 private final Cache <Id , Object > edgesCache ;
7586
76- private EventListener storeEventListener ;
7787 private EventListener cacheEventListener ;
7888 private CacheListenerHolder holder ;
89+ private StoreListenerHolder storeHolder ;
7990
8091 public CachedGraphTransaction (HugeGraphParams graph , BackendStore store ) {
8192 super (graph , store );
@@ -135,7 +146,7 @@ private void listenChanges() {
135146 Set <String > storeEvents = ImmutableSet .of (Events .STORE_INIT ,
136147 Events .STORE_CLEAR ,
137148 Events .STORE_TRUNCATE );
138- this . storeEventListener = event -> {
149+ EventListener storeListener = event -> {
139150 if (storeEvents .contains (event .name ())) {
140151 LOG .debug ("Graph {} clear graph cache on event '{}'" ,
141152 this .graph (), event .name ());
@@ -144,9 +155,24 @@ private void listenChanges() {
144155 }
145156 return false ;
146157 };
147- if (storeEventListenStatus .putIfAbsent (this .params ().spaceGraphName (), true ) == null ) {
148- this .store ().provider ().listen (this .storeEventListener );
149- }
158+ BackendStoreProvider provider = this .store ().provider ();
159+ String graphName = this .params ().spaceGraphName ();
160+ StoreListenerHolder storeAcquired = STORE_EVENT_LISTENERS .compute (
161+ graphName , (key , existing ) -> {
162+ if (existing == null || existing .provider != provider ) {
163+ // Graph close/reopen creates a new provider for the
164+ // same graph name; replace the stale holder. Old
165+ // transactions skip decrement via identity check.
166+ if (existing != null ) {
167+ existing .provider .unlisten (existing .listener );
168+ }
169+ provider .listen (storeListener );
170+ return new StoreListenerHolder (storeListener , provider );
171+ }
172+ existing .refCount ++;
173+ return existing ;
174+ });
175+ this .storeHolder = storeAcquired ;
150176
151177 // Listen cache event: "cache"(invalid cache item)
152178 EventListener listener = event -> {
@@ -196,7 +222,6 @@ private void listenChanges() {
196222 return false ;
197223 };
198224 EventHub graphEventHub = this .params ().graphEventHub ();
199- String graphName = this .params ().spaceGraphName ();
200225 CacheListenerHolder acquired = GRAPH_CACHE_EVENT_LISTENERS .compute (
201226 graphName , (key , existing ) -> {
202227 if (existing == null || existing .hub != graphEventHub ) {
@@ -235,14 +260,20 @@ private void unlistenChanges() {
235260 this .holder = null ;
236261 this .cacheEventListener = null ;
237262 }
238- // TODO (follow-up): storeEventListenStatus has the same owner-first
239- // close bug this PR fixes for GRAPH_CACHE_EVENT_LISTENERS. A non-owner
240- // transaction can remove the tracking entry, unlisten its own
241- // never-registered storeEventListener as a no-op, and leave the
242- // original store listener registered but untracked. Apply the same
243- // ref-counted holder pattern in a follow-up PR.
244- if (storeEventListenStatus .remove (graphName ) != null ) {
245- this .store ().provider ().unlisten (this .storeEventListener );
263+ StoreListenerHolder storeOurs = this .storeHolder ;
264+ if (storeOurs != null ) {
265+ STORE_EVENT_LISTENERS .compute (graphName , (key , existing ) -> {
266+ if (existing == null || existing != storeOurs ) {
267+ return existing ;
268+ }
269+ existing .refCount --;
270+ if (existing .refCount == 0 ) {
271+ existing .provider .unlisten (existing .listener );
272+ return null ;
273+ }
274+ return existing ;
275+ });
276+ this .storeHolder = null ;
246277 }
247278 }
248279
@@ -476,4 +507,27 @@ public void removeIndex(IndexLabel indexLabel) {
476507 }
477508 }
478509 }
510+
511+ /*
512+ * Listener lifetime must cover all active transactions for the graph.
513+ * The holder is removed from the registry and unregistered from the
514+ * BackendStoreProvider only when the last transaction releases it.
515+ * Mirror of CacheListenerHolder for the store event path.
516+ */
517+ private static final class StoreListenerHolder {
518+
519+ final EventListener listener ;
520+ final BackendStoreProvider provider ;
521+ // Must only be read or written inside ConcurrentMap.compute() for the
522+ // enclosing registry; ConcurrentHashMap.compute() serialises per-key
523+ // access.
524+ int refCount ;
525+
526+ StoreListenerHolder (EventListener listener ,
527+ BackendStoreProvider provider ) {
528+ this .listener = listener ;
529+ this .provider = provider ;
530+ this .refCount = 1 ;
531+ }
532+ }
479533}
0 commit comments