fix(binding-mcp): hold cache lifecycle lock for binding lifetime

McpProxyCacheManager previously released the cache lifecycle lock the
moment initial hydrate completed (in onCacheReady), opening a window
where a different worker's reconnect retry would acquire the lock and
also open its own lifecycle stream to upstream. The result was
multiple workers redundantly subscribing to the upstream SSE for the
same binding — wasted resources, and harder to reason about which
worker drives TTL refreshes.

Move releaseLifecycle out of onCacheReady. The lifecycle lock is
acquired once at binding attach (or after lifecycle abort + reconnect)
and held until detach / engine shutdown. Loser workers keep retrying
the acquire; their attempts only succeed if the holder dies and the
lock TTL expires (or releases explicitly on detach). This matches MCP
semantics: one worker per binding owns the upstream SSE; others serve
their agents from the shared cache populated via the store-watch
propagation path.

Migrate the per-kind and lifecycle locks in McpProxyCache from
putIfAbsent/delete to lock/unlock from PR #1790. Ownership-checked
unlock means a worker that never acquired the lock cannot accidentally
release another worker's lock (which the old unauthenticated delete
allowed). Token state is held on McpListCache and McpProxyCache; null
token short-circuits release as a no-op.

Engine TestStoreHandler updated to share watchers and locks per
storeConfig.id (mirroring how entries are already shared via
TestStoreContext.supplyEntries). The watcher record carries the
registering worker's signaler so cross-worker notify dispatches
listener invocations onto the registering worker's I/O thread,
matching the contract documented on StoreHandler. Without these
fixes cross-worker watch propagation either didn't fire at all or
fired on the wrong thread.

https://claude.ai/code/session_01Gx5yC2CuFd54Fyoy7kL3qg

Author: claude

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/cache/McpProxyCache.java

@@ -44,7 +44,6 @@ public final class McpProxyCache
     private static final String STORE_KEY_RESOURCES = "resources";
     private static final String STORE_KEY_PROMPTS = "prompts";
     private static final String STORE_LOCK_SUFFIX = ".lock";
-    private static final String STORE_LOCK_VALUE = "1";
     private static final String STORE_LOCK_KEY_TOOLS = STORE_KEY_TOOLS + STORE_LOCK_SUFFIX;
     private static final String STORE_LOCK_KEY_RESOURCES = STORE_KEY_RESOURCES + STORE_LOCK_SUFFIX;
     private static final String STORE_LOCK_KEY_PROMPTS = STORE_KEY_PROMPTS + STORE_LOCK_SUFFIX;
@@ -67,6 +66,7 @@ public final class McpProxyCache
     private final CRC32 crc32 = new CRC32();
 
     boolean populated;
+    private String lifecycleLockToken;
 
     Runnable onReady;
     public OnSettled onSettled;
@@ -142,14 +142,26 @@ public void register(
     void acquireLifecycle(
         Consumer<Boolean> completion)
     {
-        store.putIfAbsent(STORE_LOCK_KEY_LIFECYCLE, STORE_LOCK_VALUE, leaseTtl,
-            prior -> completion.accept(prior == null));
+        store.lock(STORE_LOCK_KEY_LIFECYCLE, leaseTtl, (k, t) ->
+        {
+            lifecycleLockToken = t;
+            completion.accept(t != null);
+        });
     }
 
     void releaseLifecycle(
         Consumer<String> completion)
     {
-        store.delete(STORE_LOCK_KEY_LIFECYCLE, completion);
+        final String token = lifecycleLockToken;
+        lifecycleLockToken = null;
+        if (token != null)
+        {
+            store.unlock(STORE_LOCK_KEY_LIFECYCLE, token, completion);
+        }
+        else
+        {
+            completion.accept(null);
+        }
     }
 
     void onPurged(
@@ -193,6 +205,7 @@ public final class McpListCache
 
         boolean populated;
         private long lastChecksum = -1L;
+        private String lockToken;
 
         private McpListCache(
             int kind,
@@ -231,14 +244,26 @@ public void put(
         public void acquire(
             Consumer<Boolean> completion)
         {
-            store.putIfAbsent(storeLockKey, STORE_LOCK_VALUE, leaseTtl,
-                prior -> completion.accept(prior == null));
+            store.lock(storeLockKey, leaseTtl, (k, t) ->
+            {
+                lockToken = t;
+                completion.accept(t != null);
+            });
         }
 
         public void release(
             Consumer<String> completion)
         {
-            store.delete(storeLockKey, completion);
+            final String token = lockToken;
+            lockToken = null;
+            if (token != null)
+            {
+                store.unlock(storeLockKey, token, completion);
+            }
+            else
+            {
+                completion.accept(null);
+            }
         }
 
         public Closeable watch(

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/cache/McpProxyCacheManager.java

@@ -177,7 +177,6 @@ private void onCacheReady()
         {
             return;
         }
-        cache.releaseLifecycle(k -> {});
         scheduleRefresh();
     }
 

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/store/TestLockEntry.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity 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 io.aklivity.zilla.runtime.engine.test.internal.store;
+
+record TestLockEntry(
+    String token,
+    long expiresAt)
+{
+}

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/store/TestStore.java

@@ -16,6 +16,7 @@
 package io.aklivity.zilla.runtime.engine.test.internal.store;
 
 import java.net.URL;
+import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -28,11 +29,15 @@ public final class TestStore implements Store
     public static final String NAME = "test";
 
     private final ConcurrentMap<Long, ConcurrentMap<String, String>> storage;
+    private final ConcurrentMap<Long, ConcurrentMap<String, List<TestWatcher>>> storageListeners;
+    private final ConcurrentMap<Long, ConcurrentMap<String, TestLockEntry>> storageLocks;
 
     public TestStore(
         Configuration config)
     {
         this.storage = new ConcurrentHashMap<>();
+        this.storageListeners = new ConcurrentHashMap<>();
+        this.storageLocks = new ConcurrentHashMap<>();
     }
 
     @Override
@@ -51,12 +56,24 @@ public URL type()
     public TestStoreContext supply(
         EngineContext context)
     {
-        return new TestStoreContext(context, this::acquireEntries);
+        return new TestStoreContext(context, this::acquireEntries, this::acquireListeners, this::acquireLocks);
     }
 
     private ConcurrentMap<String, String> acquireEntries(
         long storeId)
     {
         return storage.computeIfAbsent(storeId, id -> new ConcurrentHashMap<>());
     }
+
+    private ConcurrentMap<String, List<TestWatcher>> acquireListeners(
+        long storeId)
+    {
+        return storageListeners.computeIfAbsent(storeId, id -> new ConcurrentHashMap<>());
+    }
+
+    private ConcurrentMap<String, TestLockEntry> acquireLocks(
+        long storeId)
+    {
+        return storageLocks.computeIfAbsent(storeId, id -> new ConcurrentHashMap<>());
+    }
 }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/store/TestStoreContext.java

@@ -15,6 +15,7 @@
  */
 package io.aklivity.zilla.runtime.engine.test.internal.store;
 
+import java.util.List;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.LongFunction;
 
@@ -28,13 +29,19 @@ public final class TestStoreContext implements StoreContext
 {
     private final Signaler signaler;
     private final LongFunction<ConcurrentMap<String, String>> supplyEntries;
+    private final LongFunction<ConcurrentMap<String, List<TestWatcher>>> supplyListeners;
+    private final LongFunction<ConcurrentMap<String, TestLockEntry>> supplyLocks;
 
     public TestStoreContext(
         EngineContext context,
-        LongFunction<ConcurrentMap<String, String>> supplyEntries)
+        LongFunction<ConcurrentMap<String, String>> supplyEntries,
+        LongFunction<ConcurrentMap<String, List<TestWatcher>>> supplyListeners,
+        LongFunction<ConcurrentMap<String, TestLockEntry>> supplyLocks)
     {
         this.signaler = context.signaler();
         this.supplyEntries = supplyEntries;
+        this.supplyListeners = supplyListeners;
+        this.supplyLocks = supplyLocks;
     }
 
     @Override
@@ -46,7 +53,9 @@ public TestStoreHandler attach(
         {
             options.entries.forEach(entries::putIfAbsent);
         }
-        return new TestStoreHandler(store, signaler, entries);
+        final ConcurrentMap<String, List<TestWatcher>> listeners = supplyListeners.apply(store.id);
+        final ConcurrentMap<String, TestLockEntry> locks = supplyLocks.apply(store.id);
+        return new TestStoreHandler(store, signaler, entries, listeners, locks);
     }
 
     @Override

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/store/TestStoreHandler.java

@@ -20,7 +20,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.BiConsumer;
@@ -33,19 +32,21 @@
 public final class TestStoreHandler implements StoreHandler
 {
     private final ConcurrentMap<String, String> entries;
-    private final ConcurrentMap<String, List<BiConsumer<String, String>>> listeners;
+    private final ConcurrentMap<String, List<TestWatcher>> listeners;
     private final ConcurrentMap<String, TestLockEntry> locks;
     private final Signaler signaler;
 
     public TestStoreHandler(
         StoreConfig store,
         Signaler signaler,
-        ConcurrentMap<String, String> entries)
+        ConcurrentMap<String, String> entries,
+        ConcurrentMap<String, List<TestWatcher>> listeners,
+        ConcurrentMap<String, TestLockEntry> locks)
     {
         this.entries = Objects.requireNonNull(entries);
         this.signaler = Objects.requireNonNull(signaler);
-        this.listeners = new ConcurrentHashMap<>();
-        this.locks = new ConcurrentHashMap<>();
+        this.listeners = Objects.requireNonNull(listeners);
+        this.locks = Objects.requireNonNull(locks);
     }
 
     @Override
@@ -121,7 +122,7 @@ public void lock(
         final String token = UUID.randomUUID().toString();
         final TestLockEntry candidate = new TestLockEntry(token, expiresAt);
         TestLockEntry existing = locks.putIfAbsent(key, candidate);
-        if (existing != null && existing.expiresAt <= now)
+        if (existing != null && existing.expiresAt() <= now)
         {
             existing = locks.replace(key, existing, candidate) ? null : locks.get(key);
         }
@@ -138,14 +139,14 @@ public void unlock(
         final long now = System.currentTimeMillis();
         final TestLockEntry current = locks.get(key);
         final String result;
-        if (current != null && current.expiresAt > now && current.token.equals(token))
+        if (current != null && current.expiresAt() > now && current.token().equals(token))
         {
             locks.remove(key, current);
             result = token;
         }
         else
         {
-            if (current != null && current.expiresAt <= now)
+            if (current != null && current.expiresAt() <= now)
             {
                 locks.remove(key, current);
             }
@@ -159,14 +160,15 @@ public Closeable watch(
         String key,
         BiConsumer<String, String> listener)
     {
-        final List<BiConsumer<String, String>> list = listeners.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>());
-        list.add(listener);
+        final TestWatcher watcher = new TestWatcher(listener, signaler);
+        final List<TestWatcher> list = listeners.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>());
+        list.add(watcher);
         return () ->
         {
-            final List<BiConsumer<String, String>> current = listeners.get(key);
+            final List<TestWatcher> current = listeners.get(key);
             if (current != null)
             {
-                current.remove(listener);
+                current.remove(watcher);
             }
         };
     }
@@ -175,13 +177,13 @@ private void notifyListeners(
         String key,
         String value)
     {
-        final List<BiConsumer<String, String>> list = listeners.get(key);
+        final List<TestWatcher> list = listeners.get(key);
         if (list != null && !list.isEmpty())
         {
             final long now = System.currentTimeMillis();
-            for (BiConsumer<String, String> listener : list)
+            for (TestWatcher w : list)
             {
-                signaler.signalAt(now, 0, ignored -> listener.accept(key, value));
+                w.signaler().signalAt(now, 0, ignored -> w.listener().accept(key, value));
             }
         }
     }
@@ -192,10 +194,4 @@ private void defer(
         // contract: callback fires strictly later than the call, on the caller's I/O thread
         signaler.signalAt(System.currentTimeMillis(), 0, ignored -> task.run());
     }
-
-    private record TestLockEntry(
-        String token,
-        long expiresAt)
-    {
-    }
 }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/store/TestWatcher.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity 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 io.aklivity.zilla.runtime.engine.test.internal.store;
+
+import java.util.function.BiConsumer;
+
+import io.aklivity.zilla.runtime.engine.concurrent.Signaler;
+
+record TestWatcher(
+    BiConsumer<String, String> listener,
+    Signaler signaler)
+{
+}